症状

1つのノードに対して、4種類の非同期ワーカーがそれぞれ別のフィールドを書き込む構成でした。サムネイル生成、AI要約、スコア算出、解析結果。どれも重い処理で、1件あたり15〜60秒かかります。

ある日から、サムネイルを再生成するとスコアが消えるようになりました。逆もあります。片方を実行すると、もう片方の結果が空に戻る。

エラーは出ません。両方のワーカーが「成功」でログを残していました。

発端は「並列化した」こと

それまでは汎用の drush cron が全キューを単一プロセスで順番に処理していました。重い処理が詰まるので、対象のキューだけ2分ごとの専用 cron に切り出して並列化しました。

事故はその直後から始まりました。つまりバグは前からコードに入っていて、逐次実行だったから顔を出していなかっただけです。

素直な後勝ちレース

各ワーカーの処理はこういう形をしていました。

public function processItem($data) {
  $node = Node::load($data['nid']);   // ① 先にロード

  $result = $this->doHeavyWork($node); // ② 15〜60秒かかる

  $node->set('field_something', $result);
  $node->save();                       // ③ エンティティ全体を保存
}

$node->save() は、そのフィールドだけを UPDATE するわけではありません。$node が持っている全フィールドを書き戻します。

ワーカーAが①でロードしてから③で保存するまでの間に、ワーカーBが別のフィールドを書き込んで保存していたら、ワーカーAの $node はBの書き込みを知りません。Aの③はBの結果を「Aがロードした時点の古い値」で上書きします。

処理が15〜60秒かかるので、この窓は非常に広い。逐次実行では窓が重ならなかったので、並列化した瞬間に顕在化しました。

ロックを入れても直らなかった

対策は明らかに見えました。保存区間をロックで直列化すればいい。

$lock = \Drupal::lock();
if ($lock->acquire($lockName, 60)) {
  $node = $storage->load($nid);   // ロックの中で読み直す
  $node->set('field_something', $result);
  $node->save();
  $lock->release($lockName);
}

ロックを取り、その中で最新のノードを読み直してから保存する。これで直列化されるはずです。

直りませんでした。 消える頻度は下がりましたが、依然として消えます。

真因: エンティティの静的キャッシュがロックを無効化する

問題は $storage->load($nid) です。

Drupal のエンティティストレージは、同一リクエスト（同一プロセス）内でロードしたエンティティを静的キャッシュに保持します。同じIDを2回 load() すると、2回目はDBを見ずにキャッシュ上のオブジェクトを返します。

そしてこのワーカーは、processItem() の冒頭①で既に同じノードを load() しています。重い処理に渡すために必要だったからです。

つまりロックの中の load() は、①でロードした古いエンティティをそのまま返していました。DBには他のワーカーが書いた新しい値が入っているのに、それを読まずに、15〜60秒前のスナップショットに対してフィールドをセットして保存する。ロックは保存を直列化しましたが、直列化された各保存が古いデータを書いていたわけです。

ロックが効いていないのではなく、ロックが守っている中身が古い。これが「頻度は下がるが消える」の説明です。

対策: ロックの中で静的キャッシュを捨ててから読む

resetCache() を挟みます。

$storage = \Drupal::entityTypeManager()->getStorage('node');
// 同一ワーカープロセスは冒頭で重い処理用にノードをロード済みで、その古い
// エンティティが静的キャッシュに残る。resetCache を挟まないと load() は
// 古いキャッシュを返し、別ワーカーが DB に書いたフィールドを古い値で
// 上書き消去してしまう。必ず DB から読み直す。
$storage->resetCache([$nid]);
$node = $storage->load($nid);

これで load() はDBへ問い合わせに行きます。ロック保持中なので他ワーカーの保存と直列化されていて、読んだ値が最新であることが保証されます。

全体の形: 重い処理をロックの外に出す

最終的に、この一連を1つのヘルパーに閉じました。ポイントは重い処理を呼び出し側に残し、ロック区間を「再読込 → 値のセット → 保存」だけにすることです。

final class NodeSaveCoordinator {

  public static function lockedUpdate(int $nid, callable $apply): bool {
    $lock     = \Drupal::lock();
    $lockName = 'node_save:' . $nid;
    $acquired = FALSE;

    // ベストエフォートでロック取得（最大 ~30 秒待つ）。ロックバックエンドが
    // 使えない環境でも後段の「再読込→save」で続行する。
    try {
      for ($i = 0; $i < 6; $i++) {
        if ($lock->acquire($lockName, 60)) {
          $acquired = TRUE;
          break;
        }
        $lock->wait($lockName, 5);
      }
    }
    catch (\Throwable $e) {
      // ロック不可でも続行（再読込が衝突窓を最小化する）。
    }

    try {
      $storage = \Drupal::entityTypeManager()->getStorage('node');
      $storage->resetCache([$nid]);
      $node = $storage->load($nid);
      if (!$node instanceof NodeInterface) {
        return FALSE;
      }
      $apply($node);
      $node->save();
      return TRUE;
    }
    finally {
      if ($acquired) {
        try {
          $lock->release($lockName);
        }
        catch (\Throwable $e) {
          // release 失敗は無視（ロックは TTL で自然解放される）。
        }
      }
    }
  }
}

呼び出し側はこうなります。

public function processItem($data) {
  $node = Node::load($data['nid']);
  $result = $this->doHeavyWork($node);   // ロックの外（15〜60秒）

  NodeSaveCoordinator::lockedUpdate($data['nid'], function ($fresh) use ($result) {
    $fresh->set('field_something', $result);   // $fresh は再読込した最新ノード
  });
}

$apply に渡るのは再読込した最新のノードです。呼び出し側が最初にロードした $node は重い処理の入力にのみ使い、保存には使いません。これで別ワーカーが直前に書いたフィールドが保持されます。

設計上の細かい判断を3点だけ。

ロックはベストエフォートにした。 取得を6回試して（wait 5秒 × 最大30秒）、取れなくても続行します。ロックが取れないことを致命的な失敗にすると、キューが詰まって全体が止まります。取れなくても resetCache() + 再読込があるので衝突窓は極小になります。

acquire() のTTLは60秒。 ロック区間には重い処理が含まれないので、本来は1秒未満で終わります。TTLはプロセスが死んだ時の自然解放のための保険です。

release() は finally で、失敗は握りつぶす。 解放できなくてもTTLで消えます。ここで例外を投げると保存成功が失敗に化けます。

一般化するとどうなるか

この事故の教訓は、Drupal に限らず「ORM のアイデンティティマップ」を持つフレームワーク全般に効きます。

ロックを取っただけでは、ロックの中で読む値が新しいとは限りません。 同一プロセスが既にそのオブジェクトを読んでいれば、キャッシュ層がDBへの問い合わせを省略します。ロックは「他プロセスとの直列化」を担保しますが、「自プロセスのキャッシュの鮮度」は担保しません。

そして厄介なのは、逐次実行では絶対に再現しないことです。並列化はパフォーマンス改善のつもりで入れますが、それまで窓が重ならなかっただけのレースを一斉に顕在化させます。並列度を上げる変更は、機能追加ではなく既存コードの前提を変える変更として扱ったほうがいいです。

まとめ

  • $entity->save() はセットしたフィールドだけでなく、そのエンティティが持つ全フィールドを書き戻す。 ロードから保存までが長いと、その間の他プロセスの書き込みを古い値で上書きする
  • 重い処理を挟むワーカーは、この窓が15〜60秒級になる。 逐次実行だと窓が重ならず潜在バグのまま残る
  • 並列化はレースを作るのではなく、既にあったレースを顕在化させる。 並列度を上げる変更は既存コードの前提を変える変更として扱う
  • ロックの中で load() しても、同一プロセスが既にロード済みなら静的キャッシュが古いエンティティを返す。 ロックは直列化するが鮮度は保証しない
  • $storage->resetCache([$id]) をロックの中・load() の前に置く。 これでDBから読み直される
  • ロック区間は「再読込 → セット → 保存」だけにし、重い処理は外に出す。 ロック保持時間を短く保つ
  • ロックはベストエフォートで良い。 取得失敗を致命傷にするとキューが止まる。release() は finally で、失敗は握りつぶす（TTLで解放される）

この記事は、事業・Webサイトの売買プラットフォーム RIKKA M&A を開発する中で得た知見です。RIKKA M&A は GitHub リポジトリを解析する技術デューデリジェンスを備え、Drupal 11 / PHP / LLM で構築・運用しています。

サイト・アプリ・AIプロダクトM&Aのプラットフォーム RIKKA M&A