RxJSのforkJoinが値を出さずに終わる条件をレビューで見抜く

画面の初期化で複数のAPIを束ねるとき、forkJoin は素直な選択肢になる。全部揃ってから描画したい、という要件にそのまま対応するからだ。

ただし forkJoin の契約は「全部揃うまで待つ」ではない。正確には次のようになる。

全入力が最低1回emitして、かつ complete したとき、各入力の最後の値をまとめて1回だけemitする。

「最低1回emitして」という条件が付いている点が要になる。ここを読み落とした実装は、特定の条件でだけ画面が初期化されないという形で壊れる。

空のままcompleteする入力

次のコードは、3本のAPIを束ねて画面を初期化している。

export class DashboardComponent implements OnInit {
  private readonly api = inject(DashboardApi);

  ngOnInit(): void {
    forkJoin({
      profile: this.api.fetchProfile(),
      summary: this.api.fetchSummary(),
      notice: this.api.fetchLatestNotice(),
    }).subscribe(({ profile, summary, notice }) => {
      this.render(profile, summary, notice);
@Reviewer
`fetchLatestNotice()` は該当がないとき値を出さずにcompleteします。その場合 `forkJoin` 全体が何も出さずに終わるため、`render` が一度も呼ばれません。
    });
  }
}

fetchLatestNotice() の実装が次のようになっていたとする。

fetchLatestNotice(): Observable<Notice> {
  return this.http.get<Notice[]>('/api/notices').pipe(
    mergeMap(notices => notices),
    take(1),
  );
}

お知らせが1件以上あるときは、先頭の1件を出して complete する。件数が0のときは、mergeMap が何も流さないまま complete し、take(1) もそのまま complete する。値は一度も出ない。

このとき forkJoin は待ち続けるのではなく、その時点で値を出さずに complete する。subscribe の第1引数は呼ばれず、render も走らない。エラーも出ないので、ログを見ても手がかりがない。

空を素通しさせているのは take(1) の契約でもある。take(1) は値が来れば1件取得し、値が来ないまま source が complete しても正常終了する。条件に合う最初の1件が必須であれば first(predicate) を使い、該当がないまま complete したときに EmptyError にする選択もある。

@Reviewer: `fetchLatestNotice()` は件数0のとき値を出さずにcompleteします。`forkJoin` は入力がひとつでも空completeすると全体が値を出さずに終わるため、お知らせが無い日だけダッシュボードが描画されません。

空completeの扱いをどこに置くか

直し方は、forkJoin の側ではなく各入力の内側にある。値が出ないことを異常として扱わないなら、空の場合の値を決めてしまえばよい。

fetchLatestNotice(): Observable<Notice | null> {
  return this.http.get<Notice[]>('/api/notices').pipe(
    map(notices => notices[0] ?? null),
  );
}

HttpClient の応答は1回emitして complete するため、この形なら件数が0でも null が流れる。forkJoin の入力として成立し、呼び出し側は notice が null かどうかで分岐すればよい。

既存のObservableに手を入れにくい場合は、forkJoin へ渡す直前に defaultIfEmpty を挟む手もある。

forkJoin({
  profile: this.api.fetchProfile(),
  summary: this.api.fetchSummary(),
  notice: this.api.fetchLatestNotice().pipe(defaultIfEmpty(null)),
}).subscribe(/* ... */);

どちらを選ぶかは、空を「この関数の正常な結果」と見るか「この画面での扱い」と見るかで決まる。前者なら関数側、後者なら呼び出し側に置くほうが、後から読んだときに意図が追いやすい。

1本が失敗したとき他の入力はどうなるか

空complete と並んで見落とされるのが、エラー時のふるまいである。forkJoin は入力のひとつが error になると全体も error になり、そのとき他の入力からは unsubscribe する。

部分的な失敗を許容したい画面では、この挙動が要件と合わない。お知らせの取得だけ失敗しても、プロフィールと集計は表示したい、という要件はよくある。

type NoticeResult = { ok: boolean; notice: Notice | null };

const emptyNotice: NoticeResult = { ok: true, notice: null };
const failedNotice: NoticeResult = { ok: false, notice: null };

forkJoin({
  profile: this.api.fetchProfile(),
  summary: this.api.fetchSummary(),
  notice: this.api.fetchLatestNotice().pipe(
    map((notice): NoticeResult => ({ ok: true, notice })),
    catchError(() => of(failedNotice)),
    defaultIfEmpty(emptyNotice),
  ),
}).subscribe(/* ... */);

要点は、catchError を forkJoin の外側ではなく各入力の内側へ置くことにある。外側に置いても部分的な結果は受け取れない。forkJoin は全入力が complete した後に結合した値を一度だけemitする契約なので、error へ分岐した時点で結合値そのものが作られないからだ。先に complete していた入力があっても変わらない。

@Reviewer: お知らせの取得が失敗すると `forkJoin` 全体がerrorになり、成功したプロフィールと集計も破棄されます。部分表示を許容する画面であれば、各入力の内側で捕捉して値へ変換してください。

合流Operatorごとのemitの条件

forkJoin を選ぶ場面では、他の合流Operatorが候補に上がっていることも多い。いつemitするかだけを並べると、選択の基準がはっきりする。

画面初期化のように「一度だけ揃えば終わり」なら forkJoin が合う。値が更新され続け、最新の組み合わせを追い続けたいなら combineLatest になる。ここで combineLatest を選んだ場合、complete を待たない代わりに、全ソースが一度emitするまでは何も出ないという別の条件が付く。

なお、Store の状態同士を組み合わせる場面では、combineLatest ではなく Selector の合成を検討することになる。これは合流Operatorの契約とは別の論点なので、NgRx Store を扱う回で改めて取り上げる。

レビュー観点チェックリスト

おわりに

forkJoin が値を出さずに終わる不具合は、再現条件が「お知らせが0件の日」のようなデータ依存になりやすい。テスト環境では常にデータが入っているため、レビューを抜けると本番まで残りやすい。

コードを読むときに見るべきなのは forkJoin そのものではなく、渡されている入力ひとつひとつが「必ず値を出して終わるか」である。ここが確認できていれば、forkJoin は要件どおりに動く。