途中で失敗したエージェントの副作用を巻き戻す — 補償トランザクション設計の実装メモ

type StepResult<T> = { ok: true; value: T } | { ok: false; error: Error };
 
interface SagaStep<Ctx> {
  name: string;
  // 順方向。戻り値は後続ステップとロールバック判断のためにcontextへ格納する
  do: (ctx: Ctx) => Promise<unknown>;
  // 巻き戻し。do が成功した場合のみ、逆順に呼ばれる
  compensate: (ctx: Ctx) => Promise<void>;
  // この地点を越えたら後退できない（後述のpoint of no return）
  pivot?: boolean;
}
 
interface SagaReport {
  status: "completed" | "compensated" | "stuck";
  completedSteps: string[];
  failedStep?: string;
  compensationFailures: string[];
}
 
async function runSaga<Ctx>(steps: SagaStep<Ctx>[], ctx: Ctx): Promise<SagaReport> {
  const done: SagaStep<Ctx>[] = [];
  let crossedPivot = false;
 
  for (const step of steps) {
    try {
      await step.do(ctx);
      done.push(step);
      if (step.pivot) crossedPivot = true;
    } catch (error) {
      // 部分失敗が確定した瞬間
      if (crossedPivot) {
        // 後退できない地点を越えている → 前進回復に切り替える(後述)
        return {
          status: "stuck",
          completedSteps: done.map((s) => s.name),
          failedStep: step.name,
          compensationFailures: [],
        };
      }
      return await compensate(done, ctx, step.name);
    }
  }
  return { status: "completed", completedSteps: done.map((s) => s.name), compensationFailures: [] };
}

async function compensate<Ctx>(
  done: SagaStep<Ctx>[],
  ctx: Ctx,
  failedStep: string,
): Promise<SagaReport> {
  const compensationFailures: string[] = [];
  // 逆順 = 最後に成功したものから取り消す
  for (const step of [...done].reverse()) {
    try {
      await step.compensate(ctx);
    } catch (e) {
      // 補償の失敗は止めない。記録して人間に上げる対象にする
      compensationFailures.push(step.name);
    }
  }
  return {
    status: "compensated",
    completedSteps: done.map((s) => s.name),
    failedStep,
    compensationFailures,
  };
}

const releaseSaga: SagaStep<ReleaseCtx>[] = [
  {
    name: "admob-mediation",
    do: async (ctx) => {
      ctx.admobSnapshot = await admob.getMediationConfig(ctx.appId); // 巻き戻し用に現状を保存
      await admob.updateMediationConfig(ctx.appId, ctx.newMediation);
    },
    compensate: async (ctx) => {
      if (ctx.admobSnapshot) await admob.updateMediationConfig(ctx.appId, ctx.admobSnapshot);
    },
  },
  {
    name: "appstore-metadata",
    do: async (ctx) => {
      ctx.metaSnapshot = await appstore.getMetadata(ctx.appId);
      await appstore.updateMetadata(ctx.appId, ctx.newMetadata);
    },
    compensate: async (ctx) => {
      if (ctx.metaSnapshot) await appstore.updateMetadata(ctx.appId, ctx.metaSnapshot);
    },
  },
  {
    name: "remote-config-flag",
    // フラグ点灯はユーザーに新機能が見え始める = 実質的な不可逆点
    pivot: true,
    do: async (ctx) => {
      await remoteConfig.setFlag(ctx.appId, "new_feature", true);
    },
    compensate: async (ctx) => {
      await remoteConfig.setFlag(ctx.appId, "new_feature", false); // 冪等
    },
  },
];

// do/compensate を計測ラッパで包み、観測点を一本化する
function observed<Ctx>(step: SagaStep<Ctx>): SagaStep<Ctx> {
  return {
    ...step,
    do: async (ctx) => {
      const t = Date.now();
      try {
        const r = await step.do(ctx);
        log.info({ saga: "release", step: step.name, phase: "do", ms: Date.now() - t });
        return r;
      } catch (e) {
        log.error({ saga: "release", step: step.name, phase: "do", failed: true });
        throw e;
      }
    },
    compensate: async (ctx) => {
      log.warn({ saga: "release", step: step.name, phase: "compensate", started: true });
      await step.compensate(ctx);
      log.warn({ saga: "release", step: step.name, phase: "compensate", done: true });
    },
  };
}