Merge pull request #1785 from j5ik2o/remote-graceful-flush

feat(remote): add graceful flush lifecycle

Author: j5ik2o

docs/gap-analysis/remote-gap-analysis.md

@@ -1,6 +1,6 @@
 # remote モジュール ギャップ分析
 
-更新日: 2026-05-14 (13th edition / payload serialization 接続後)
+更新日: 2026-05-14 (14th edition / remote graceful flush 接続後)
 
 ## 比較スコープ定義
 
@@ -44,27 +44,27 @@ raw declaration count は Scala / Java / JVM 固有 API を含む参考値であ
 
 remote は address primitives、failure detector、association state、wire PDU、TCP transport shell、resolve cache、remote `ActorRef` materialization、actor-core serialization registry backed payload の outbound / inbound delivery まで実装済みである。
 
-一方で、残ギャップは compression table application、remote deployment、`AddressTerminated` integration、flush lifecycle に集中している。remote DeathWatch の watch / unwatch / notification delivery と ACK/NACK redelivery は `remote-reliable-deathwatch` で接続済みである。
+一方で、残ギャップは compression table application、remote deployment、`AddressTerminated` integration に集中している。remote DeathWatch の watch / unwatch / notification delivery、ACK/NACK redelivery、shutdown / DeathWatch 前 flush lifecycle は接続済みである。
 
 | 指標 | 値 |
 |------|-----|
 | Pekko 固定スコープ対象概念 | 75 |
-| fraktor-rs 固定スコープ対応概念 | 70 |
-| 固定スコープ概念カバレッジ | 70/75 (93.3%) |
+| fraktor-rs 固定スコープ対応概念 | 72 |
+| 固定スコープ概念カバレッジ | 72/75 (96.0%) |
 | raw Pekko public type declarations | 361（Scala / Java、protobuf 除外） |
 | raw Pekko `def` declarations | 1594 |
-| raw fraktor public type declarations | 86（`remote-core`: 70 / `remote-adaptor-std`: 16） |
-| raw fraktor public method declarations | 352（`remote-core`: 312 / `remote-adaptor-std`: 40） |
-| hard / medium / easy / trivial gap | 5 / 0 / 0 / 0 |
+| raw fraktor public type declarations | 87（`remote-core`: 73 / `remote-adaptor-std`: 14、簡易再計測） |
+| raw fraktor public method declarations | 346（`remote-core`: 328 / `remote-adaptor-std`: 18、簡易再計測） |
+| hard / medium / easy / trivial gap | 3 / 0 / 0 / 0 |
 
 `todo!()` / `unimplemented!()` / `panic!("not implemented")` と production code 上の明示 TODO は remote core / adaptor から検出されない。`modules/remote-core/src/wire/primitives.rs:12` の header placeholder は encode 時の長さ埋め戻しであり、未実装ギャップには分類しない。
 
 ## 層別カバレッジ
 
 | 層 | Pekko 対応範囲 | fraktor-rs 現状 | 評価 |
 |----|----------------|-----------------|------|
-| core | address、unique address、association、wire PDU、failure detector、watcher state、provider contract、typed config | `modules/remote-core/src/` に整理済み。no_std 側の状態機械と PDU は揃っている | 公開 primitive は強い。compression / deployment / flush は残る |
-| std / adaptor | TCP listener/client、association 実行系、remoting lifecycle、inbound dispatch、reconnect/backoff、serialized payload delivery、watcher task | `TcpRemoteTransport`、`run_inbound_dispatch`、`run_remote_with_delivery`、`StdRemoteActorRefProvider`、watcher task は存在 | bind / handshake / reconnect / quarantine filter / actor-core serializer backed payload delivery / remote DeathWatch は動く。flush driver はない |
+| core | address、unique address、association、wire PDU、failure detector、watcher state、provider contract、typed config | `modules/remote-core/src/` に整理済み。no_std 側の状態機械と PDU は揃っている | 公開 primitive は強い。compression / deployment は残る |
+| std / adaptor | TCP listener/client、association 実行系、remoting lifecycle、inbound dispatch、reconnect/backoff、serialized payload delivery、watcher task | `TcpRemoteTransport`、`run_inbound_dispatch`、`run_remote_with_delivery`、`StdRemoteActorRefProvider`、watcher task は存在 | bind / handshake / reconnect / quarantine filter / actor-core serializer backed payload delivery / remote DeathWatch / flush driver は動く |
 | actor-core integration | serialization registry、ActorRefProvider、DeathWatch、event stream、routing/deploy | misc serializer、scheme provider lookup、remote `ActorRef` materialization、routee expansion、registered payload remote send、remote DeathWatch 通知は接続済み | remote deployment と `AddressTerminated` integration が残る |
 
 ## カテゴリ別ギャップ
@@ -111,14 +111,9 @@ remote は address primitives、failure detector、association state、wire PDU
 
 `RemotingLifecycleState`, `Remote`, `RemoteShared`, `EventPublisher`, `RemoteLogMarker`, `RemoteInstrument`, `RemotingFlightRecorder`, `RemoteAuthoritySnapshot`、主要 `RemoteConfig` builder は実装済み。`bind_hostname` / `bind_port` / `inbound_lanes` / `outbound_lanes` / `maximum_frame_size` / `buffer_pool_size` / `untrusted_mode` / log toggle / outbound queue / remove-quarantined / outbound restart budget / inbound restart budget / large-message destinations / compression config は現行コードで確認済み。
 
-### 8. Reliability / lifecycle adaptor ✅ 実装済み 2/4 (50%)
+### 8. Reliability / lifecycle adaptor ✅ 実装済み 4/4 (100%)
 
-`InboundQuarantineCheck` 相当の quarantine handling と connection loss recovery は実装済み。shutdown / DeathWatch 前 flush は残る。
-
-| Pekko API / 契約 | Pekko参照 | fraktor対応 | 実装先層 | 難易度 | 備考 |
-|------------------|-----------|-------------|----------|--------|------|
-| `FlushOnShutdown` | `artery/FlushOnShutdown.scala:30`, `artery/FlushOnShutdown.scala:52`, `artery/FlushOnShutdown.scala:79` | 未対応 | std/extension_installer + std/association | hard | `shutdown_flush_timeout` 設定はあるが、association に termination hint / flush frame を送り ack を待つ driver がない |
-| `FlushBeforeDeathWatchNotification` | `artery/FlushBeforeDeathWatchNotification.scala:33`, `artery/FlushBeforeDeathWatchNotification.scala:65`, `artery/FlushBeforeDeathWatchNotification.scala:85` | 未対応 | std/watcher + std/association | hard | DeathWatch 通知前に対象 association を flush する契約がない |
+`InboundQuarantineCheck` 相当の quarantine handling、connection loss recovery、`FlushOnShutdown` 相当の shutdown flush wait、`FlushBeforeDeathWatchNotification` 相当の DeathWatch notification 前 flush gate は実装済み。
 
 ### 9. Internal helpers / cache ✅ 実装済み 2/2 (100%)
 
@@ -147,12 +142,11 @@ remote は address primitives、failure detector、association state、wire PDU
 
 ## 内部モジュール構造ギャップ
 
-固定スコープ概念カバレッジは 93.3% で 80% を超えるため、公開 API の残ギャップを実装する上での構造差分も記録する。残る API gap はすべて `hard` であり、次の構造ギャップが実装順序を制約している。
+固定スコープ概念カバレッジは 96.0% で 80% を超えるため、公開 API の残ギャップを実装する上での構造差分も記録する。残る API gap はすべて `hard` であり、次の構造ギャップが実装順序を制約している。
 
 | 構造ギャップ | Pekko側の根拠 | fraktor-rs側の現状 | 推奨アクション | 難易度 | 緊急度 | 備考 |
 |-------------|---------------|--------------------|----------------|--------|--------|------|
 | compression table 所有者不足 | `artery/compress/InboundCompressions.scala:39`, `artery/Codecs.scala:260` | `RemoteCompressionConfig` は設定値のみで、table state / advertisement owner がない | core に table state、std に advertisement timer / control delivery を配置 | hard | medium | serializer registry 接続後に実装しやすい |
-| flush protocol の PDU / driver 不足 | `FlushOnShutdown.scala:52`, `FlushBeforeDeathWatchNotification.scala:65` | `shutdown_flush_timeout` はあるが、flush frame / ack / wait driver がない | core wire に flush control、std association に wait driver を追加 | hard | medium | DeathWatch 前 flush と shutdown flush の共通基盤 |
 | remote deployment の責務境界不足 | `RemoteActorRefProvider.scala:596`, `RemoteDaemon.scala:59` | actor-core に `RemoteScope` はあるが、remote daemon / deployment watcher がない | std provider と actor-core deployer の境界に remote create command を追加 | hard | medium | deployer と serialization の両方にまたがる |
 
 ## 実装優先度
@@ -174,13 +168,11 @@ remote は address primitives、failure detector、association state、wire PDU
 | actor-ref / manifest compression advertisement and table application | core/wire + std/transport/tcp + actor-core-kernel/serialization | 4 |
 | remote deployment daemon / `useActorOnNode` | std/provider + actor-core-kernel | 5 |
 | `AddressTerminated` integration | actor-core-kernel + std | 6 |
-| `FlushOnShutdown` | std/extension_installer + std/association | 8 |
-| `FlushBeforeDeathWatchNotification` | std/watcher + std/association | 8 |
 
 ## まとめ
 
-remote は address primitives、association state machine、ACK/NACK redelivery、failure detector + registry、typed `RemoteConfig`、TCP transport shell、inbound quarantine、restart handling、resolve cache、remote `ActorRef` materialization、remote DeathWatch、主要 misc serialization、registered payload の二ノード配送までカバー済みで、基礎部品の parity は進んでいる。
+remote は address primitives、association state machine、ACK/NACK redelivery、failure detector + registry、typed `RemoteConfig`、TCP transport shell、inbound quarantine、restart handling、resolve cache、remote `ActorRef` materialization、remote DeathWatch、shutdown / DeathWatch 前 flush、主要 misc serialization、registered payload の二ノード配送までカバー済みで、基礎部品の parity は進んでいる。
 
 Phase 1 / Phase 2 の低コストギャップは現時点ではない。parity を次に進めるには、Phase 3 のうち actor-ref / manifest compression table application を先に通すのが効果的である。
 
-主要ギャップは compression table application、`AddressTerminated` integration、remote deployment、flush lifecycle に集中している。API カバレッジは 90% を超えているため、次のボトルネックは単なる型追加ではなく、compression table / deployment / flush の接続境界にある。
+主要ギャップは compression table application、`AddressTerminated` integration、remote deployment に集中している。API カバレッジは 90% を超えているため、次のボトルネックは単なる型追加ではなく、compression table / deployment / actor-core failure notification の接続境界にある。

docs/plan/2026-05-14-remote-graceful-flush-implementation.md

@@ -0,0 +1,19 @@
+# remote-graceful-flush implementation plan
+
+## Objective
+
+Implement OpenSpec change `remote-graceful-flush` with minimal, task-tracked changes.
+
+## Plan
+
+1. Add wire-level flush control PDUs and codec coverage.
+   Verify with `fraktor-remote-core-rs` wire tests.
+2. Add core association flush session state and transport-neutral outcomes.
+   Verify duplicate ack, timeout, connection loss, and pending queue behavior with core unit tests.
+3. Add `Remote` / `RemoteShared` flush start, timer input, inbound ack/request handling, and outcome observation.
+   Verify the std adaptor can consume outcomes without exposing raw locks or association references.
+4. Add lane-targeted flush delivery to `RemoteTransport` and `TcpRemoteTransport`.
+   Verify writer-lane targeting, lane 0 handling, backpressure, inbound request routing, and ack send failures.
+5. Add std flush gate for remote-bound `DeathWatchNotification` and update `RemotingExtensionInstaller::shutdown_and_join`.
+   Verify timeout/start-failure/completion all release pending shutdown or notification paths.
+6. Run the OpenSpec-requested verification commands and update the remote gap analysis when implementation is complete.

modules/remote-adaptor-std/src/extension_installer.rs

@@ -5,6 +5,9 @@
 #[path = "extension_installer_test.rs"]
 mod tests;
 
+mod flush_gate;
 mod remoting_extension_installer;
 
+pub(crate) use flush_gate::{StdFlushGate, StdFlushNotification};
+pub(crate) use remoting_extension_installer::RemoteProviderFlushHandles;
 pub use remoting_extension_installer::RemotingExtensionInstaller;