Transactional Observation of Values

Observation モジュールに、クロージャで初期化する AsyncSequence 型 Observations を追加します。クロージャ内で参照された @Observable プロパティが変わるたびに、そのクロージャを再評価した結果を値として流す、という単純な仕組みです。

let names = Observations { person.name }

for await name in names {
  print(name)
}

このクロージャの中身は任意の式でよく、複数プロパティや条件分岐を組み合わせられます。@Observable な pet プロパティを参照する例は次のとおりです。

let greetings = Observations {
  if let pet = person.pet {
    return "Hello \(person.name) and \(pet.name)"
  } else {
    return "Hello \(person.name)"
  }
}

person.pet が nil から別の Pet に差し替わったり、その pet.name が変わったりしたときに、それぞれが追跡対象として自動的に拾われます。

型の形

Observations のシグネチャはおおむね次のとおりです。

public struct Observations<Element: Sendable, Failure: Error>: AsyncSequence, Sendable {
  public init(
    @_inheritActorContext _ emit: @escaping @isolated(any) @Sendable () throws(Failure) -> Element
  )

  public enum Iteration: Sendable {
    case next(Element)
    case finished
  }

  public static func untilFinished(
    @_inheritActorContext _ emit: @escaping @isolated(any) @Sendable () throws(Failure) -> Iteration
  ) -> Observations<Element, Failure>
}

クロージャは @isolated(any) かつ @_inheritActorContext が付いていて、Observations を生成した場所の isolation をそのまま引き継ぎます。生成が @MainActor の文脈なら、以降そのクロージャは常に @MainActor 上で実行されます。この性質のおかげで、観測される @Observable 型自体は Sendable でなくても安全に扱えます。

クロージャは typed throws なので、エラーを投げれば、現在走っているイテレーションはその Failure で終了し、以降の next() は nil を返します。

トランザクションで tearing を防ぐ

Observations の核は、「次のサスペンションポイントまで待って、整合性のとれた値を一度だけ流す」挙動です。クロージャはまず初回の next() 呼び出し時に、捕捉した isolation の上で withObservationTracking の中で実行されます。追跡中のプロパティの 最初の willSet でトランザクションが開始され、その isolation 上の次のサスペンションでクロージャが再評価されて、結果が一つの値としてイテレータに届きます。

let person = Person(firstName: "", lastName: "")
// willSet \.firstName でトランザクション開始
person.firstName = "Jane"
// willSet \.lastName ではまだトランザクションは dirty のまま
person.lastName = "Appleseed"
// 次のサスペンションで name が "Jane Appleseed" として一度だけ流れる

firstName と lastName を続けて代入してもイベントは一度しか発生せず、name が "Jane " のような中途半端な値として観測されることはありません。これは isolation による排他アクセスの保証に乗っかっているため、Sendable でない型でも安全に成り立ちます。

終了を表現する untilFinished

通常の初期化子は、クロージャの戻り値そのものを要素として流し続けるので、ローカルには終わりがありません。戻り値を Optional にしてしまうと、「本当に nil を値として流したい」場面と「もう終わり」を区別できなくなるので、終端を明示したい場合は untilFinished を使います。

let hosts = Observations.untilFinished { [weak person] in
  if let person {
    .next(person.homePage?.host)
  } else {
    .finished
  }
}

.next(value) で値を流し、.finished を返した時点でシーケンスは終了します。[weak person] で参照を持たない形にしておけば、person が解放されたタイミングで自然に終われます。

挙動のポイント

いくつかの観測の性質を押さえておくと便利です。

初回値の priming

各イテレータは、最初の next() 呼び出しで捕捉した isolation にホップしてクロージャを一度評価し、その結果を最初の要素として返します。これにより、どのイテレータでも「まずは現在の値を一つ受け取る」ところから始まります。

@MainActor
func example() async {
  let person = Person(firstName: "0", lastName: "0")
  let names = Observations { person.name }
  Task {
    try await Task.sleep(for: .seconds(2))
    person.firstName = "1"
    person.lastName  = "1"
  }
  Task { for await name in names { print("A = \(name)") } }
  Task { for await name in names { print("B = \(name)") } }
  try? await Task.sleep(for: .seconds(10))
}
// A = 0 0
// B = 0 0
// B = 1 1
// A = 1 1

生成速度がイテレーションを上回るとき

クロージャ側で変更が連続して起きたとき、イテレーション側がそれに追いつかなければ、間の値はスキップされ、最新のトランザクションだけが次の next() で観測されます。Observations は中間値をバッファしません。

@MainActor
func iterate(_ names: Observations<String, Never>) async {
  for await name in names {
    print(name)
    try? await Task.sleep(for: .seconds(0.095))
  }
}

生成が 0.1 秒ごとでイテレーションが少し遅いと、観測されるのは各トランザクションの整合した値ですが、すべての中間状態が拾われるわけではありません。整合性は保証、網羅は保証しない、というのが基本的なトレードオフです。

異なる isolation との行き来

Observations を生成した isolation とは別の isolation からイテレーションしても問題ありません。クロージャは常に生成時の isolation で実行されるので、観測対象の型は自前で Sendable でなくとも、その isolation の中に閉じ込めたまま読み取れます。

@globalActor
actor ExcplicitlyAnotherActor: GlobalActor {
  static let shared = ExcplicitlyAnotherActor()
}

@ExcplicitlyAnotherActor
func iterate(_ names: Observations<String, Never>) async {
  for await name in names { print(name) }
}

@MainActor
func example() async throws {
  let person = Person(firstName: "", lastName: "")
  // @MainActor 上でクロージャが走る
  let names = Observations { person.name }

  Task.detached { await iterate(names) }
  // 以降 MainActor 上で更新
}

複数イテレータへの同時配信

同じ Observations を複数の Task からイテレートすると、それぞれのイテレータが同じ値を同じトランザクション単位で受け取ります。余計なバッファを介さず、クロージャの評価結果が各イテレータの continuation に同時に配られる形です。ただし、AとBのどちらが先に print するかといった順序は非決定的です。

キャンセルと終了

Observations は通常の AsyncSequence のキャンセル挙動に従います。Task がキャンセルされたイテレータは nil を返して終了し、クロージャが throw した場合はその Failure でイテレーションが終わって以降は nil を返し続けます。キャンセルはイテレータごとに独立なので、ある Task をキャンセルしても他の Task のイテレーションには影響しません。

APIを公開する側への示唆

従来は、@Observable モデルの変化を外部に公開したいときに、型側で AsyncSequence プロパティ（例: trackingStates のようなもの）を手作りで用意する選択肢が目立っていました。Observations が入ると、「モデルは素直に @Observable にしておき、AsyncSequence が必要な利用者は Observations { model.foo } で自分で組み立てる」という運用がしやすくなります。値どうしが連動していて tearing を避けたいモデルや、UI 系と連携するモデルは特に、@Observable + Observations の組を優先的に検討する価値が出てきました。一方で、外部イベント由来の独立したストリームは、これまで通り AsyncSequence プロパティとして見せるほうが自然です。