外部API呼び出しを含む処理を、advisory lockではなく自作のハートビートで排他制御する

問題

時間のかかる後処理を、同じユーザーの複数のリクエストが同時に始めてしまうことがある。
PostgreSQLのadvisory lockでは、処理の間ずっとDB接続を握り続けてしまうため、この用途には向いていない。

具体例

ユーザーの操作をきっかけに、後処理を始める。
後処理の中身は「外部APIへの問い合わせ」と「DBへの保存」の2段階。
外部APIの応答には数秒から数分かかることがある。

同じユーザーが、別のタブや別の操作から同時にリクエストを送ると、後処理が同時に2つ動いてしまうことがある。

解決方法

対象（ユーザーなど）ごとに、状態を書くための行を1つ用意する。

target_id	updated_at
user-1	10:00:30
user-2	10:01:00

最初に処理を始めたリクエストが、その行に自分の担当開始時刻を記録する。
これを「担当権を取る」と呼ぶ。

担当権を取ったリクエストは、処理を続けながら、別にタイマーを動かす。
タイマーは一定間隔（例として30秒）で、同じ行の更新時刻を今の時刻に書き換える。
処理が落ちるとタイマーも止まるので、更新時刻が新しければ、処理はまだ生きているとわかる。
心臓の鼓動のように、定期的に「生きている」と知らせる合図なので、ハートビートと呼ぶ。

同じユーザーから別のリクエストが来て、同じ処理を始めようとしたときは、まずそのユーザーの行を見に行く。
最後のハートビートから一定時間（例として2分）たっていなければ、担当中とみなして何もしない。
一定時間ハートビートが止まっていたら、担当の処理が止まったとみなして、自分が担当権を奪う。
別のユーザーのリクエストは、別の行を見るので、ぶつからない。

advisory lockとは

PostgreSQLにある、アプリが決めた番号に鍵をかける仕組み。

select pg_advisory_lock(123);   -- 123番に鍵をかける
-- ここで処理をする
select pg_advisory_unlock(123); -- 鍵を外す

同じ番号に別のリクエストが鍵をかけようとすると、鍵が外れるまで待たされる。
表や行ではなく、ただの番号に鍵をかける。

「advisory」は「助言の」という意味。
DBがデータを守ってくれるわけではない。
「この番号に鍵がかかっていたら待つ」と、アプリ側で約束して使う。

実際には、番号を手で決めるのではなく、対象のIDから作ることが多い。

select pg_advisory_lock(hashtext('memory:user-1'));

hashtextは、文字列を番号に変えるPostgreSQLの関数。
同じ文字列からは、いつも同じ番号ができる。

同じユーザーへの処理は同じ番号になるので、わざとぶつかる。
別のユーザーへの処理は別の番号になるので、待たずに動く。

なぜDB接続を握り続けるのか

advisory lockは、鍵をかけた接続にひもづいている。

• 鍵をかけた接続が続いている間は、鍵が残る
• 接続が切れると、鍵も自動で外れる

そのため、外部APIの応答を何分も待つ間も、その接続を手放せない。
DB接続の数には上限があるので、これが重なると接続が足りなくなる。

advisory lockとの使い分け

	advisory lock	ハートビート
向いている処理	短い処理（DBの読み書きだけ）	長い処理（外部APIへの問い合わせなど）
DB接続	処理が終わるまで握り続ける	時刻を書くときだけ使う
処理が落ちたとき	DBが自動で鍵を外す	一定時間（例：2分）たつまで、だれも担当権を取れない

advisory lockは、落ちたときの後始末をDBに任せられるのがよいところ。
ただ、外部APIの応答を何分も待つ処理では、DB接続を握り続けてしまう。
そこで、長い処理にはハートビートを使う。

実装例（疑似コード）

担当権テーブルは、対象のID・合言葉（lease_token）・更新時刻の3列。

create table job_lease (
  target_id text primary key,
  lease_token uuid not null,
  updated_at timestamptz not null
);

lease_tokenは、担当権を取るたびに新しく作る。
担当権を奪われたあとも前の担当が動いていることがあるので、前と今の担当を見分けるために使う。

担当権は、1つのSQLで取る。
行がない、または更新時刻が古いときだけ書き込めるので、同時に来ても取れるのは片方だけになる。

async function acquireLease(targetId: string, timeoutMs: number) {
  const leaseToken = crypto.randomUUID();
  const result = await db.execute(sql`
    insert into job_lease (target_id, lease_token, updated_at)
    values (${targetId}, ${leaseToken}, now())
    on conflict (target_id) do update
      set lease_token = ${leaseToken}, updated_at = now()
      where job_lease.updated_at < now() - ${timeoutMs} * interval '1 millisecond'
  `);
  return result.rowCount > 0 ? leaseToken : null; // 1行書けたら、担当権を取れた
}

全体の流れは次のとおり。
ハートビート・保存・手放す処理は、すべてlease_tokenを条件に入れる。
合言葉が変わっていたら、担当権を奪われたということなので、何もしない。

const leaseToken = await acquireLease(userId, 2 * 60 * 1000);
if (!leaseToken) return; // ほかのリクエストが担当中

const timer = startHeartbeat(userId, leaseToken, 30 * 1000);
try {
  const data = await callExternalApi(userId); // 数秒〜数分かかる
  await saveIfStillOwner(userId, leaseToken, data);
} finally {
  clearInterval(timer); // エラーのときも、必ずタイマーを止める
  await releaseLease(userId, leaseToken);
}

ハートビート・保存・手放す処理のコード

ハートビートを送る

タイマーで一定間隔ごとに、更新時刻を今の時刻に書き換える。
lease_tokenが一致しなければ0行になるので、前の担当は、自分がもう担当ではないと気づける。

function startHeartbeat(targetId: string, leaseToken: string, intervalMs: number) {
  return setInterval(async () => {
    const result = await db.execute(sql`
      update job_lease set updated_at = now()
      where target_id = ${targetId} and lease_token = ${leaseToken}
    `);

    // 0行なら、担当権はすでに他のリクエストに奪われている
    if (result.rowCount === 0) {
      // ここでタイマーを止め、処理そのものも中断する
    }
  }, intervalMs);
}

結果を保存する

確かめてから保存するまでの間に担当権を奪われると、二重保存を防げない。
そこで、確認と保存を1つのトランザクションにまとめる。
for updateで担当権の行をロックし、保存が終わるまで、ほかのリクエストが行を更新できないようにする。

async function saveIfStillOwner(targetId: string, leaseToken: string, resultData: unknown) {
  return db.transaction(async (tx) => {
    const lease = await tx.execute(sql`
      select lease_token from job_lease
      where target_id = ${targetId}
      for update
    `);

    if (lease.rows[0]?.lease_token !== leaseToken) {
      // 担当権はもう自分のものではないので、結果は保存せず捨てる
      return false;
    }

    await tx.execute(sql`
      insert into job_result (target_id, data)
      values (${targetId}, ${resultData})
      on conflict (target_id) do update set data = excluded.data
    `);

    return true;
  });
}

担当権を手放す

処理が終わったら、lease_tokenを条件にして行を消す。
target_idだけで消すと、あとから担当権を取った別のリクエストの行まで消してしまう。

async function releaseLease(targetId: string, leaseToken: string) {
  await db.execute(sql`
    delete from job_lease
    where target_id = ${targetId} and lease_token = ${leaseToken}
  `);
}

気をつけた点

タイムアウトは、ハートビートの間隔より十分に長くする

間隔とタイムアウトが近いと、通信が少し遅れただけで、まだ動いている担当から担当権を奪ってしまう。
逆に長すぎると、本当に処理が止まったとき、引き継ぐまでに時間がかかる。
今回は、間隔を30秒、タイムアウトを2分にした（ハートビート4回分、合図がなかったら奪う）。

担当権を取れたことだけで安心しない

担当権を奪ったあとも、前の担当がまだ動いていて、結果を保存しようとすることがある。
そのため、ハートビート・保存・手放す、すべてのタイミングでlease_tokenを確かめる。
保存は、確認と保存を1つのトランザクションにまとめ、確かめた直後に奪われる隙間もなくしている。

まとめ

外部APIへの問い合わせを含む、時間の読めない処理を、同じユーザーの複数のリクエストが同時に始めてしまう問題があった。
advisory lockはDB接続を握り続ける前提の仕組みなので、この用途には向いていなかった。

対象ごとに1行用意し、時刻を書くだけのハートビート方式の担当権で解決した。
advisory lockでは、接続が切れたときにPostgreSQLが鍵を外してくれる。
ハートビートでは、「一定時間合図がなければ外れたとみなす」と、アプリ自身が決める。
仕組みはシンプルだが、担当権を外す役目をPostgreSQLからアプリに移す、という発想の転換が肝になっている。