You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Analysis: Legacy Stream Components That Can Be Refactored to GraphStage
After analyzing the codebase, here is a structured inventory of components still using the legacy Pump/InputBunch/OutputBunch/TransferPhase infrastructure that could be refactored to modern GraphStage equivalents.
The most complex legacy component. Handles bidirectional SSL/TLS encryption/decryption with a sophisticated state machine using TransferPhase patterns (bidirectional, flushingOutbound, awaitingClose, outboundClosed, inboundClosed). Uses custom TransferState implementations like ChoppingBlock for byte chunking.
Key complexity: SSL handshake state management, half-closed connection handling, buffer chopping/put-back logic, and the maxTLSIterations loop guard for unwrapping.
2. Unzip (Legacy) (High Priority, Low Complexity)
Simple fan-out that splits Tuple2/Pair elements into two output streams. Already has a modern GraphStage counterpart in UnzipWith2. The legacy version can be directly replaced.
3. FanoutProcessorImpl (High Priority)
Reactive Streams processor supporting multiple downstream subscribers. Includes SubscriptionTimeout handling. Used by BroadcastHub and similar fan-out patterns.
Eliminate Legacy Infrastructure: Each refactored component removes dependency on the Pump/InputBunch/OutputBunch/TransferPhase system, which is the old Akka Streams architecture predating GraphStage. This reduces the overall codebase maintenance burden.
Materializer Simplification: Components with special Island tags (TlsModule, SourceModule, SinkModule, ProcessorModule) require separate materialization phases. Refactoring to GraphStage means fewer islands, simpler traversal, and less special-case handling in PhasedFusingActorMaterializer.
Uniform Architecture: GraphStage is the modern, well-understood abstraction. Having all operators use the same model makes the codebase more consistent, easier to understand, and simpler for new contributors.
Better Testability: GraphStage logic (GraphStageLogic) is more testable in isolation compared to actor-based implementations that require actor system setup and message passing.
Future-Proofing: The legacy Pump infrastructure is technical debt from early Akka Streams. Removing it incrementally (one component at a time, opt-in via config like the TlsModule approach) allows safe migration without breaking changes.
Fusion Opportunities: Some legacy components (like Unzip) could potentially be fused with adjacent stages when converted to GraphStage, reducing the number of async boundaries and improving throughput. Note: TLSActor likely won't benefit from fusion (must run as its own island due to SSLEngine state management), but still benefits from uniform architecture.
Reduced Actor Overhead: Each legacy component spawns a dedicated actor. GraphStage components can be fused into a single ActorGraphInterpreter actor, reducing actor count, memory overhead, and context switching.
Based on the current apache/pekko codebase, I re-validated the scope of #2860 and would propose splitting the work into a sequence of small PRs.
Scope correction
Unzip is no longer a public GraphStage migration target: the public DSL scaladsl.Unzip is already GraphStage-based. What remains is an internal legacy impl.FanOut.Unzip, which appears unused and is a good first cleanup candidate.
Proposed PR order
PR-1: remove internal legacy impl.FanOut.Unzip
small, low-risk cleanup
if needed, add directional regression coverage around the current public Unzip behavior
PR-2: migrate the Sink.asPublisher(fanout = true) path
behavior-sensitive and TCK-sensitive, so this should stay separate from PR-1
PR-3: add a GraphStage-backed TLS path behind a safe switch
keep the existing TLS actor/island path as fallback initially
PR-4: TLS cutover and retirement of old TLS actor/island infrastructure
PR-5: final cleanup of now-unused legacy substrate
only after grep + compile + MiMa prove there are no remaining users
Notes on dependencies
Each change should be a separate PR.
If a PR depends on an earlier one, the PR description should explicitly list the required merge order.
TLS should not be the first migration PR; it is the highest-risk area because of custom materialization/island behavior, SSLEngine state handling, and compatibility concerns.
Both slices were independently cross-reviewed locally (code/RS semantics/safety/test adequacy) before opening the draft PRs.
Next up: continue with the TLS scope from #2860. The current plan is to treat TLS as its own focused migration area and keep the performance-sensitive actor/island constraints explicit during the design and review work.
*Four "reliably cancel subscriptions" tests are marked pending with detailed Scaladoc. Root cause: async boundary ordering — demand (1 inter-actor hop) consistently wins over failure (2 hops) in the TLS actor mailbox. The async boundary itself is intentional (SSLEngine thread-safety + per-connection actor isolation). See class Scaladoc in TlsGraphStageSpec for full explanation.
TLSActor is used in pekko-http. So we need to retain that class in its current form for a period of a few weeks. So we could add its replacement but keep the old code for a few weeks to keep our builds working. We can update pekko-http but let's not break it and force a panicked update.
Current-main recheck for #2860 (Pekko a55a008fdf, 2026-10-29). This supersedes parts of the older remaining-work list.
The TLS GraphStage implementation and the default-engine migration are separate milestones. The new engine is implemented and has substantial regression coverage, but pekko.stream.materializer.tls.engine still defaults to legacy-actor. Review and maintainer feedback identified two regressions: accepted plaintext lost on cancellation (#3727), and buffer-allocation/growth failures (#3728). Both PRs have been revised to address the review feedback.
Remove fanout publisher sink wrapper #3176 removed FanoutPublisherSink. Current Sink.asPublisher(true) directly composes Flow[T].async with the GraphStage bridge. This is no longer a future rewrite.
Review against the original TLSActor state machine
The comparison uses the actual TransferState algebra, not just matching phase names: A || B completes when both complete, while A && B completes when either completes (Transfer.scala:114).
Original phase
GraphStage counterpart
Contract to preserve
bidirectional
BidirectionalPhase
Inbound first, then outbound; engine NEED_WRAP can progress without application data.
flushingOutbound
FlushingOutboundPhase
Flush protocol output subject to transport demand; handle SSLException without stranding shutdown.
awaitingClose
AwaitingClosePhase
Continue consuming peer close records, discarding application output.
outboundClosed
OutboundClosedPhase
Continue inbound processing and required handshake/close writes.
inboundClosed
InboundClosedPhase
Preserve ignore-cancel behavior while processing the remaining outbound side.
completedPhase / pumpFinished
CompletedPhase / StageCompletionPendingFlag
Cancel inputs, complete outputs, and finish already-pending output before stopping.
The original states are at TLSActor.scala:233, and the explicit port predicates are at TlsGraphStage.scala:186. The main new proof obligations are batching, delayed flushes, output slots, and pooled buffers. #3324 is an example of why copying the old phases alone was insufficient.
I also checked an early cipher-output cancellation hypothesis with both implementations. With two Source.maybe inputs, a server engine waiting for input, Sink.cancelled on cipher output, and Sink.ignore on plaintext output, the legacy and GraphStage paths behaved alike in all four async/no-async combinations. That experiment does not establish a migration regression, so it is not being presented as one or used to justify a redundant spec.
Correction following PR review: my earlier classification of the public TcpSpec failure as merely a closing-policy assumption was wrong. GraphStage had already accepted the final plaintext into its timer-delayed batch, then closed outbound on plaintext-output cancellation before encrypting it. The default four-argument bindAndHandleWithTls test returned hell instead of hello.
#3727 now fixes the engine and restores that original test unchanged. Cancellation stops new plaintext input, drains already-accepted data for the established session under transport backpressure, and then closes. It does not start queued renegotiation or wait indefinitely for a new handshake. Focused tests cover both cancel-sensitive policies, initial-handshake cancellation, queued renegotiation, handshake transitions during draining, transport cancellation, and delayed cancellation. Legacy TLS is unchanged. This is why comparing phase predicates alone was insufficient: the new batching layer introduced an extra state obligation.
Confirmed gap: TLS allocation depends on the TCP pool size
TlsGraphStage.scala:144 acquires three transport buffers from the TCP pool. acquireTransportBuffer:682 throws if a fixed-size pooled buffer is smaller than SSLSession.getPacketBufferSize. Legacy TLS allocates its own buffers independently of TCP.
On JDK 21 with -Dpekko.io.tcp.direct-buffer-size=8192:
TlsSpec: 113 passed.
TlsGraphStageSpec, filtered to SingleBytes: 8 failed, covering TLS 1.2/1.3 and in-memory/TCP communication patterns.
All eight fail at startup: TLS packet buffer requires 16709 bytes but pekko.io.tcp.direct-buffer-size is 8192.
The differential run used the compiled ScalaTest classes. An equivalent focused sbt reproduction is:
There is also a confirmed runtime failure even when the configured buffer fits the initial packet size. growTransportOutBuffer:578 asks the same fixed-size pool for twice the current capacity, which cannot succeed.
A second differential probe used real JDK 21 TLS 1.2 engines, completed the handshake, and round-tripped an initial 2,048-byte payload with a 17,000-byte pool. Holding the batching timer then allowed a 1,024-byte wrap to retain 1,053 ciphertext bytes. The next 2,048-byte write encountered only 15,947 free bytes, below the provider's 16,709-byte requirement, so wrap returned BUFFER_OVERFLOW. GraphStage attempted to obtain 34,000 bytes from the 17,000-byte pool and failed; legacy TLS delivered all 5,120 application bytes. The wrapper only observed real engine results; it did not fabricate the overflow.
Following review, #3728 flushes existing ciphertext before retrying an overflow; it grows only when an empty output buffer is too small. It verifies acquired capacity, returns only correctly sized direct buffers to the pool, and resizes inbound packet/application buffers if session requirements increase. Batching uses actual output capacity and is refreshed after growth.
Focused tests cover TLS 1.2/1.3 with an 8 KiB pool, genuine empty-buffer growth, all pooled buffers' sizes/directness/unique identities after termination, flush-before-grow without an allocation, and deterministic inbound session-size increases. The latter explicitly assert that overflow/underflow occurred. Existing deferred-close tests remain. Heap fallback has a per-connection allocation cost; the documentation now states the allocation and batching tradeoffs. A separate TLS pool or normal memory-policy change remains separate performance work.
The default pool uses 128 KiB buffers, so the three held buffers cost approximately 384 KiB of direct memory per live TLS stage, before application buffers. #3019 proposed heap/on-demand buffers but was closed without merging. That work remains relevant; measure native memory as well as heap allocation and throughput.
Downstream status
Current pekko-http main (127528261) has no source references to TLSActor, TlsModule, TlsUtils, or TlsGraphStage; HTTP and HTTP/2 use the public TLS(...) API. The earlier direct-dependency warning should not be repeated as an unverified current blocker.
pekko-http already runs both TLS engines across JDK 17/21/25 and Scala 2.13/3. All 12 jobs in the 2026-10-28 nightly succeeded. The JDK 21/Scala 2.13/graph-stage log used Pekko 2.0.0-M4+190-2c035294-SNAPSHOT: this is existing integration evidence, not a downstream rerun against today's exact HEAD. Published downstream versions and other consumers still need checking before class removal.
The goal is to complete as many beneficial GraphStage execution-path migrations as possible in 2.0.0. Moving execution and deleting old classes are separate decisions. Keep TLSActor, TlsModule, and their supporting compatibility/fallback classes in 2.0.0; reassess retirement in 2.1.0 or 2.2.0 after downstream and binary-compatibility evidence. This is a reassessment point, not a promise to delete them in either release.
Priority
Work
Boundary / acceptance criteria
P0
TLS buffer compatibility and accepted-plaintext cancellation fixes (#3728, #3727), then a separate default-switch PR targeting 2.0.0
Preserve both engines' contracts; cross-JDK TLS, HTTP/HTTP2 and Artery validation; native-memory and latency evidence. Keep legacy-actor as a transition fallback.
P1
Revisit TLS transport-buffer memory policy for 2.0.0 in a separate change
Normal pooled buffers still retain approximately 384 KiB of direct memory per live stage. Reassess the ideas in #3019 with real engine overflow/underflow, batching, throughput, latency, and live-connection memory measurements.
Investigate within 2.0.0
Flow.fromProcessor(Mat) / ProcessorModulePhase and remaining Reactive Streams adapters
Preserve one factory call per materialization, the original materialized value, subscription/cancellation ordering, and direct passthrough where useful. ProcessorModulePhase itself does not create an actor. A wrapper must demonstrate a benefit before replacement; do not automatically defer a proven useful rewrite to 2.1.0.
Retain optimized path
Sink.cancelled / CancelSink
Withdraw the earlier P1 GraphStage-conversion proposal. The current subscriber cancels directly in onSubscribe; preserve this fast cancellation path. A future fusion-aware implementation needs topology-specific evidence, not a blanket replacement.
Retain compatibility in 2.0.0
TLSActor, TlsModule, FanIn / FanOut, and the buffering/Pump/Transfer support chain
Do not delete classes merely because the preferred execution path has migrated. Shared companions/protocols are still used by the legacy fallback, and internal visibility is not a binary-compatibility exemption. Review retirement in 2.1.0/2.2.0.
PublisherSource, SubscriberSource, SubscriberSink, and single-subscriber PublisherSink are Reactive Streams boundary adapters, not ordinary actor operators. PublisherSink deliberately uses VirtualPublisher to delay subscription. They remain candidates for evidence-based optimization during 2.0.0; merely wrapping them in GraphStages is not sufficient acceptance evidence.
Sink.cancelled performance recheck
CancelSink creates a small CancellingSubscriber, whose onSubscribe directly invokes Subscription.cancel(). It has no actor or Pump state machine of its own.
For Source.fromPublisher(p).to(Sink.cancelled), SinkModulePhase wires the publisher directly to that subscriber. A temporary GraphStage prototype that calls completeStage() in preStart instead runs cancellation through the interpreter and its input boundary. A deterministic latch-based experiment blocked the selected stream dispatcher: the existing implementation still cancelled on the calling thread; the prototype cancelled only after the dispatcher was released. Both requested zero elements.
There is an important qualification: the current materializer unconditionally creates a default island, invokes its onIslandReady, and can create an empty interpreter actor even for the old direct-adapter topology. The verified advantage is cancellation does not wait for actor scheduling, not zero actors or zero background work for the entire materialization.
The same prototype was compared with the unchanged implementation in a temporary JMH 1.37 harness on JDK 17: two forks per case, three 1-second warmups, four 1-second measurements, one benchmark thread, and GC profiling. It covered direct Publisher, Publisher plus map(identity), and a GraphStage source. The measurement ends when cancellation (or source postStop) is observed; it does not await complete actor teardown. Direct-Publisher cancellation was slower in the prototype; the pure GraphStage-source case showed potential fusion/allocation benefits. The mapped-Publisher results and some fork timings had substantial variance. Cancellation notification also precedes final actor cleanup, and the 3-second warmup does not establish steady state for delayed subscription-timeout cleanup. These short, shared-machine runs therefore do not establish stable performance ratios or universal throughput/allocation superiority.
For an already fused stream, a GraphStage sink can avoid the existing ActorOutputBoundary. That is a topology-specific opportunity, not a justification for losing the direct cancellation path. No production conversion, duplicate spec, or Sink.cancelled PR has been created.
Useful TLS follow-ups before changing the default:
Extend batching, fragmented-record, slow-consumer and timer/cancellation edge cases to TLS 1.3; the dedicated edge-case fixture currently fixes TLS 1.2, although the shared suite covers both protocols.
Keep Artery TLS coverage with GraphStage explicitly selected in the cutover gate; the six cases run here passed on JDK 21, rather than constituting a cross-JDK Artery matrix.
Retain public-API selection coverage in isolated JVMs/CI after Preserve accepted TLS plaintext on output cancellation #3727. Selection is currently a classloader-global val using ConfigFactory.load(), not each materializer's config; the selection unit tests exercise precedence helpers. The local TCP and Artery runs here explicitly selected GraphStage.
Improve benchmark scope: warmRoundTrip:148 creates a fresh client/server pair for every 1,000-record invocation, despite the class comment saying engines are reused. Add an actual long-lived-connection fixture, TLS 1.3, latency, and live-connection memory measurements. Historical throughput numbers alone do not establish those properties.
In 2.1.0 or 2.2.0, reassess whether downstream validation and binary compatibility permit retiring TlsModule, its materializer island/phase, and TLSActor, then pruning the unused buffering/Pump/Transfer chain. Until then, keep them available. The still-used TlsUtils.applySessionParameters is shared with the GraphStage; any file move must preserve its binary name and method shape.
Do not blanket-delete ActorSubscriberMessage or ActorPublisher.scala: current GraphStage internals use the former; the latter owns shutdown sentinels used by the GraphInterpreter, fanout bridge, subscriber management, and hubs. Actor-ref sources, stream refs, TCP and file/stream I/O adapters are already GraphStages. Interpreter/supervisor actors are runtime infrastructure, outside this operator migration.
Validation
Baseline small-pool differential: 113 legacy passes versus 8 GraphStage failures at startup. A separate real-engine overflow probe also reproduced the fixed-size-pool failure.
Preserve accepted TLS plaintext on output cancellation #3727: three new accepted-data cases failed before the fix. The revised branch passes 257 TLS regression tests on JDK 21 and the original four-argument TCP TLS test with GraphStage selected. All 10 focused close/cancellation cases pass on JDK 17, 21, and 25.
Fix TLS GraphStage buffer allocation and growth #3728: flush/growth and increased-packet tests exposed failures on the previous PR head. The revised branch passes 256 TLS regression tests on JDK 21. All 9 focused buffer/close cases pass on JDK 17, 21, and 25; session-growth tests use fixed initial sizes and explicit overflow/underflow assertions.
Earlier real-engine validation delivered all 5,120 bytes after overflow; 8 small-pool shared communication cases and 6 Artery TLS cases passed with GraphStage selected. These are supporting runs, not a claim that every integration case was rerun on both latest PR heads.
Native scalafmt, header generation, and git diff --check passed. The local sortImports attempt hit a scalameta linkage error; unrelated edits were removed.
Full validation, MiMa, and documentation builds are delegated to remote CI as requested. Both PR branches were force-updated after review, so old green checks are not evidence for the new heads; the new CI runs are in progress.
All eight inline review threads have been addressed and resolved. No default-engine change or legacy-class removal is included.
The Sink.cancelled investigation used temporary probe/benchmark files only. Production sources remain unchanged.
These fixes, the default-engine switch, normal buffer-memory policy, and eventual legacy-class retirement remain separate changes with their own acceptance evidence.
The reproduced hell/hello failure does not show that TLS had accepted the final o before observing plaintext-output cancellation. In every failing diagnostic run, the server-side event order was accept(h), accept(e), accept(l), accept(l), cancel, accept(o).
The statement that Preserve accepted TLS plaintext on output cancellation #3727 "restores the original four-argument test" was incorrect. TcpSpec.scala, its four-argument call, closing policy, and hello assertion are unchanged in the PR. The referenced 31f6681c00 is superseded by 205155160a.
The current PR description also needs a qualification: only h/e/l/l had been accepted before cancellation. A late onPush(o) was observed afterwards on both revisions; saying that o never entered the operator would be inaccurate.
Result
#3727 fixes the separate loss of plaintext already accepted at cancellation; it does not eliminate the original TcpSpec failure. The observed echo failure is a cancellation-versus-delivery race, outside that drain guarantee. These logs establish the event order at TLS, not the exact upstream buffer responsible. The follow-up is #3729.
The drain also applies when plaintext-output cancellation is observed before peer transport-input completion. If transport-input completion is observed first, the existing inbound-close contract discards accepted plaintext, matching the legacy actor engine.
Tests
On 2026-10-29, using OpenJDK 25.0.2, macOS arm64, Scala 2.13.18, and the unchanged TLS 1.3 TcpSpec case:
All failures returned hell, expected hello. Each repeat ran the original ScalaTest case in a fresh JVM using sbt-exported test classpath and JVM options; stream-tests/Test/fork = true was confirmed, and GraphStage was explicitly selected by system property.
A separate diagnostic batch ran 20 times per engine per revision. GraphStage failed 11/20 (55%) on main and 12/20 (60%) on the PR, always with the event order above. Successful GraphStage runs accepted hello before cancellation. Both legacy batches passed 20/20 and logged hello at receipt and ChoppingBlock.chopInto. Diagnostic changes were local only and removed.
These are shared-workstation observations, not comparable failure-probability estimates: the one-minute load ranged from 5.76–6.72 during the main baseline and 19.40–34.44 during the PR baseline; logging can also affect scheduling. Earlier individual passes do not establish that this intermittent case was fixed.
References
Refs #2860, #3727, and #3729. Buffer allocation/growth remains separate in #3728.
Activity
He-Pin commentedon Apr 18, 2026
Analysis: Legacy Stream Components That Can Be Refactored to GraphStage
After analyzing the codebase, here is a structured inventory of components still using the legacy
Pump/InputBunch/OutputBunch/TransferPhaseinfrastructure that could be refactored to modernGraphStageequivalents.Component Analysis Table
impl/io/TLSActor.scalaActor + PumpInputBunch(2),OutputBunch(2),TransferPhase, customTransferStateBidiShapeimpl/FanOut.scala:328FanOut + PumpOutputBunch,BatchingInputBuffer,TransferPhaseFanOutShape2impl/FanoutProcessor.scala:135ActorProcessorImpl + PumpBatchingInputBuffer,FanoutOutputs,TransferPhaseFlowShapeimpl/Sinks.scala:114SinkModule→ createsFanoutProcessorImplSinkShapeimpl/FanIn.scalaActor + PumpInputBunch,SimpleOutputsimpl/FanOut.scala:273Actor + PumpOutputBunch,BatchingInputBufferimpl/ActorProcessor.scala:272Actor + PumpBatchingInputBuffer,SimpleOutputsFlowShapeimpl/io/TlsModule.scalaAtomicModule+ custom IslandTlsModuleIslandTag, spawnsTLSActorBidiShapeDetailed Breakdown
1. TLSActor (Highest Priority)
The most complex legacy component. Handles bidirectional SSL/TLS encryption/decryption with a sophisticated state machine using
TransferPhasepatterns (bidirectional,flushingOutbound,awaitingClose,outboundClosed,inboundClosed). Uses customTransferStateimplementations likeChoppingBlockfor byte chunking.Key complexity: SSL handshake state management, half-closed connection handling, buffer chopping/put-back logic, and the
maxTLSIterationsloop guard for unwrapping.2. Unzip (Legacy) (High Priority, Low Complexity)
Simple fan-out that splits
Tuple2/Pairelements into two output streams. Already has a modernGraphStagecounterpart inUnzipWith2. The legacy version can be directly replaced.3. FanoutProcessorImpl (High Priority)
Reactive Streams processor supporting multiple downstream subscribers. Includes
SubscriptionTimeouthandling. Used byBroadcastHuband similar fan-out patterns.Legacy Infrastructure Dependency Map
Advantages of Refactoring to GraphStage
Eliminate Legacy Infrastructure: Each refactored component removes dependency on the
Pump/InputBunch/OutputBunch/TransferPhasesystem, which is the old Akka Streams architecture predating GraphStage. This reduces the overall codebase maintenance burden.Materializer Simplification: Components with special
Islandtags (TlsModule,SourceModule,SinkModule,ProcessorModule) require separate materialization phases. Refactoring to GraphStage means fewer islands, simpler traversal, and less special-case handling inPhasedFusingActorMaterializer.Uniform Architecture: GraphStage is the modern, well-understood abstraction. Having all operators use the same model makes the codebase more consistent, easier to understand, and simpler for new contributors.
Better Testability: GraphStage logic (
GraphStageLogic) is more testable in isolation compared to actor-based implementations that require actor system setup and message passing.Future-Proofing: The legacy Pump infrastructure is technical debt from early Akka Streams. Removing it incrementally (one component at a time, opt-in via config like the TlsModule approach) allows safe migration without breaking changes.
Fusion Opportunities: Some legacy components (like Unzip) could potentially be fused with adjacent stages when converted to GraphStage, reducing the number of async boundaries and improving throughput. Note: TLSActor likely won't benefit from fusion (must run as its own island due to SSLEngine state management), but still benefits from uniform architecture.
Reduced Actor Overhead: Each legacy component spawns a dedicated actor. GraphStage components can be fused into a single
ActorGraphInterpreteractor, reducing actor count, memory overhead, and context switching.Recommended Order
ProcessorModuleIslandTag.He-Pin commentedon Apr 18, 2026
Based on the current
apache/pekkocodebase, I re-validated the scope of #2860 and would propose splitting the work into a sequence of small PRs.Scope correction
Unzipis no longer a public GraphStage migration target: the public DSLscaladsl.Unzipis already GraphStage-based. What remains is an internal legacyimpl.FanOut.Unzip, which appears unused and is a good first cleanup candidate.Proposed PR order
impl.FanOut.UnzipUnzipbehaviorSink.asPublisher(fanout = true)pathFanoutPublisherSink/FanoutProcessorImpl/ required replacement plumbingNotes on dependencies
Validation focus
GraphUnzipSpec, stream compile, MiMaFanoutProcessorSpec,TimeoutsSpec, fanout-related TCK, MiMaTlsSpec,TcpSpec,MaterializerStateSpec, remoteTlsTcpSpec, relevant artery TLS coverage, MiMaI have not started code changes yet in this thread; this comment is only to align the execution plan before taking the first slice.
chore(stream): remove internal FanOut Unzip
feat(stream): replace fanout publisher runtime with GraphStage bridge
He-Pin commentedon Apr 18, 2026
Progress update:
Two review-complete draft PR slices are now open for issue #2860:
PR chore(stream): remove internal FanOut Unzip #2873 - chore(stream): remove internal FanOut Unzip
org.apache.pekko.stream.impl.Unzipimplementation fromFanOut.scala2.0.xMiMa excludesstream/test:compile,stream/mimaReportBinaryIssues,GraphUnzipSpecPR feat(stream): replace fanout publisher runtime with GraphStage bridge #2874 - feat(stream): replace fanout publisher runtime with GraphStage bridge
FanoutProcessorImplruntime behindSink.asPublisher(fanout = true)withFanoutPublisherBridgeStageFanoutProcessor.scalaFanoutPublisherBehaviorSpec2.0.xMiMa excludesstream/test:compile,stream/mimaReportBinaryIssues,FanoutPublisherBehaviorSpec,TimeoutsSpec,FlowSpec,SinkSpec,SinkTest,FanoutPublisherTestBoth slices were independently cross-reviewed locally (code/RS semantics/safety/test adequacy) before opening the draft PRs.
Next up: continue with the TLS scope from #2860. The current plan is to treat TLS as its own focused migration area and keep the performance-sensitive actor/island constraints explicit during the design and review work.
chore(stream): format fanout publisher bridge for CI
chore(stream): remove internal FanOut Unzip (#2873)
feat(stream): add GraphStage-backed TLS path behind feature switch (#…
He-Pin commentedon Apr 19, 2026
PR-3 Progress Update
PR #2877 has been opened as a draft: feat(stream): add GraphStage-backed TLS path behind feature switch
What is in PR-3
A new
TlsGraphStage(pureBidiGraphStageimplementation of TLS) is added behind a runtime feature switch:pekko.stream.materializer.tls.use-legacy-actor = off-Dpekko.stream.materializer.tls.use-legacy-actor=falseTLSActorpath unchangedTest status
TlsSpec(legacy path)TlsGraphStageSpec(GraphStage path)MaterializerStateSpec*Four "reliably cancel subscriptions" tests are marked
pendingwith detailed Scaladoc. Root cause: async boundary ordering — demand (1 inter-actor hop) consistently wins over failure (2 hops) in the TLS actor mailbox. The async boundary itself is intentional (SSLEngine thread-safety + per-connection actor isolation). See class Scaladoc inTlsGraphStageSpecfor full explanation.PRs so far
impl.FanOut.Unzip)Sink.asPublisher(fanout=true)to GraphStage)Coming up
TLSActor/TlsModuleretirement (after PR-3 is reviewed/merged)FanIn.scala/FanOut.scalasubstrateHe-Pin commentedon Apr 19, 2026
pjfanning commentedon Apr 19, 2026
TLSActor is used in pekko-http. So we need to retain that class in its current form for a period of a few weeks. So we could add its replacement but keep the old code for a few weeks to keep our builds working. We can update pekko-http but let's not break it and force a panicked update.
He-Pin commentedon Apr 19, 2026
I think.we may need keep it in 2.20.0 and remove it in 2.1.0？
feat(stream): add clean-room GraphStage TLS implementation (TlsGraphS…
feat(stream): add clean-room GraphStage TLS implementation (TlsGraphS…
feat(stream): replace fanout publisher runtime with GraphStage bridge (…
14 remaining items
He-Pin commentedon Oct 28, 2026
Current-main recheck for #2860 (Pekko
a55a008fdf, 2026-10-29). This supersedes parts of the older remaining-work list.The TLS GraphStage implementation and the default-engine migration are separate milestones. The new engine is implemented and has substantial regression coverage, but
pekko.stream.materializer.tls.enginestill defaults tolegacy-actor. Review and maintainer feedback identified two regressions: accepted plaintext lost on cancellation (#3727), and buffer-allocation/growth failures (#3728). Both PRs have been revised to address the review feedback.Already completed
FanoutPublisherBridgeStage; fix(stream): preserve abrupt fanout publisher shutdown signal #2910 fixed abrupt-shutdown signaling.ActorProcessor,ActorProcessorImpl, andExposedPublisherReceivetypes.ActorProcessor.scalastill contains buffering helpers used by legacy TLS.FanoutPublisherSink. CurrentSink.asPublisher(true)directly composesFlow[T].asyncwith the GraphStage bridge. This is no longer a future rewrite.Review against the original TLSActor state machine
The comparison uses the actual
TransferStatealgebra, not just matching phase names:A || Bcompletes when both complete, whileA && Bcompletes when either completes (Transfer.scala:114).bidirectionalBidirectionalPhaseflushingOutboundFlushingOutboundPhaseawaitingCloseAwaitingClosePhaseoutboundClosedOutboundClosedPhaseinboundClosedInboundClosedPhasecompletedPhase/pumpFinishedCompletedPhase/StageCompletionPendingFlagThe original states are at TLSActor.scala:233, and the explicit port predicates are at TlsGraphStage.scala:186. The main new proof obligations are batching, delayed flushes, output slots, and pooled buffers. #3324 is an example of why copying the old phases alone was insufficient.
I also checked an early cipher-output cancellation hypothesis with both implementations. With two
Source.maybeinputs, a server engine waiting for input,Sink.cancelledon cipher output, andSink.ignoreon plaintext output, the legacy and GraphStage paths behaved alike in all four async/no-async combinations. That experiment does not establish a migration regression, so it is not being presented as one or used to justify a redundant spec.Correction following PR review: my earlier classification of the public
TcpSpecfailure as merely a closing-policy assumption was wrong. GraphStage had already accepted the final plaintext into its timer-delayed batch, then closed outbound on plaintext-output cancellation before encrypting it. The default four-argumentbindAndHandleWithTlstest returnedhellinstead ofhello.#3727 now fixes the engine and restores that original test unchanged. Cancellation stops new plaintext input, drains already-accepted data for the established session under transport backpressure, and then closes. It does not start queued renegotiation or wait indefinitely for a new handshake. Focused tests cover both cancel-sensitive policies, initial-handshake cancellation, queued renegotiation, handshake transitions during draining, transport cancellation, and delayed cancellation. Legacy TLS is unchanged. This is why comparing phase predicates alone was insufficient: the new batching layer introduced an extra state obligation.
Confirmed gap: TLS allocation depends on the TCP pool size
TlsGraphStage.scala:144 acquires three transport buffers from the TCP pool. acquireTransportBuffer:682 throws if a fixed-size pooled buffer is smaller than
SSLSession.getPacketBufferSize. Legacy TLS allocates its own buffers independently of TCP.On JDK 21 with
-Dpekko.io.tcp.direct-buffer-size=8192:TlsSpec: 113 passed.TlsGraphStageSpec, filtered toSingleBytes: 8 failed, covering TLS 1.2/1.3 and in-memory/TCP communication patterns.TLS packet buffer requires 16709 bytes but pekko.io.tcp.direct-buffer-size is 8192.The differential run used the compiled ScalaTest classes. An equivalent focused sbt reproduction is:
There is also a confirmed runtime failure even when the configured buffer fits the initial packet size. growTransportOutBuffer:578 asks the same fixed-size pool for twice the current capacity, which cannot succeed.
A second differential probe used real JDK 21 TLS 1.2 engines, completed the handshake, and round-tripped an initial 2,048-byte payload with a 17,000-byte pool. Holding the batching timer then allowed a 1,024-byte wrap to retain 1,053 ciphertext bytes. The next 2,048-byte write encountered only 15,947 free bytes, below the provider's 16,709-byte requirement, so
wrapreturnedBUFFER_OVERFLOW. GraphStage attempted to obtain 34,000 bytes from the 17,000-byte pool and failed; legacy TLS delivered all 5,120 application bytes. The wrapper only observed real engine results; it did not fabricate the overflow.Following review, #3728 flushes existing ciphertext before retrying an overflow; it grows only when an empty output buffer is too small. It verifies acquired capacity, returns only correctly sized direct buffers to the pool, and resizes inbound packet/application buffers if session requirements increase. Batching uses actual output capacity and is refreshed after growth.
Focused tests cover TLS 1.2/1.3 with an 8 KiB pool, genuine empty-buffer growth, all pooled buffers' sizes/directness/unique identities after termination, flush-before-grow without an allocation, and deterministic inbound session-size increases. The latter explicitly assert that overflow/underflow occurred. Existing deferred-close tests remain. Heap fallback has a per-connection allocation cost; the documentation now states the allocation and batching tradeoffs. A separate TLS pool or normal memory-policy change remains separate performance work.
The default pool uses 128 KiB buffers, so the three held buffers cost approximately 384 KiB of direct memory per live TLS stage, before application buffers. #3019 proposed heap/on-demand buffers but was closed without merging. That work remains relevant; measure native memory as well as heap allocation and throughput.
Downstream status
Current pekko-http main (
127528261) has no source references toTLSActor,TlsModule,TlsUtils, orTlsGraphStage; HTTP and HTTP/2 use the publicTLS(...)API. The earlier direct-dependency warning should not be repeated as an unverified current blocker.pekko-http already runs both TLS engines across JDK 17/21/25 and Scala 2.13/3. All 12 jobs in the 2026-10-28 nightly succeeded. The JDK 21/Scala 2.13/graph-stage log used Pekko
2.0.0-M4+190-2c035294-SNAPSHOT: this is existing integration evidence, not a downstream rerun against today's exact HEAD. Published downstream versions and other consumers still need checking before class removal.2.0.0 scope: maximize useful execution-path rewrites, retain compatibility classes
The goal is to complete as many beneficial GraphStage execution-path migrations as possible in 2.0.0. Moving execution and deleting old classes are separate decisions. Keep
TLSActor,TlsModule, and their supporting compatibility/fallback classes in 2.0.0; reassess retirement in 2.1.0 or 2.2.0 after downstream and binary-compatibility evidence. This is a reassessment point, not a promise to delete them in either release.legacy-actoras a transition fallback.Flow.fromProcessor(Mat)/ProcessorModulePhaseand remaining Reactive Streams adaptersProcessorModulePhaseitself does not create an actor. A wrapper must demonstrate a benefit before replacement; do not automatically defer a proven useful rewrite to 2.1.0.Sink.cancelled/CancelSinkonSubscribe; preserve this fast cancellation path. A future fusion-aware implementation needs topology-specific evidence, not a blanket replacement.TLSActor,TlsModule,FanIn/FanOut, and the buffering/Pump/Transfer support chainPublisherSource,SubscriberSource,SubscriberSink, and single-subscriberPublisherSinkare Reactive Streams boundary adapters, not ordinary actor operators.PublisherSinkdeliberately usesVirtualPublisherto delay subscription. They remain candidates for evidence-based optimization during 2.0.0; merely wrapping them in GraphStages is not sufficient acceptance evidence.Sink.cancelled performance recheck
CancelSink creates a small CancellingSubscriber, whose
onSubscribedirectly invokesSubscription.cancel(). It has no actor or Pump state machine of its own.For
Source.fromPublisher(p).to(Sink.cancelled), SinkModulePhase wires the publisher directly to that subscriber. A temporary GraphStage prototype that callscompleteStage()inpreStartinstead runs cancellation through the interpreter and its input boundary. A deterministic latch-based experiment blocked the selected stream dispatcher: the existing implementation still cancelled on the calling thread; the prototype cancelled only after the dispatcher was released. Both requested zero elements.There is an important qualification: the current materializer unconditionally creates a default island, invokes its onIslandReady, and can create an empty interpreter actor even for the old direct-adapter topology. The verified advantage is cancellation does not wait for actor scheduling, not zero actors or zero background work for the entire materialization.
The same prototype was compared with the unchanged implementation in a temporary JMH 1.37 harness on JDK 17: two forks per case, three 1-second warmups, four 1-second measurements, one benchmark thread, and GC profiling. It covered direct Publisher, Publisher plus
map(identity), and a GraphStage source. The measurement ends when cancellation (or sourcepostStop) is observed; it does not await complete actor teardown. Direct-Publisher cancellation was slower in the prototype; the pure GraphStage-source case showed potential fusion/allocation benefits. The mapped-Publisher results and some fork timings had substantial variance. Cancellation notification also precedes final actor cleanup, and the 3-second warmup does not establish steady state for delayed subscription-timeout cleanup. These short, shared-machine runs therefore do not establish stable performance ratios or universal throughput/allocation superiority.For an already fused stream, a GraphStage sink can avoid the existing ActorOutputBoundary. That is a topology-specific opportunity, not a justification for losing the direct cancellation path. No production conversion, duplicate spec, or Sink.cancelled PR has been created.
Useful TLS follow-ups before changing the default:
valusingConfigFactory.load(), not each materializer's config; the selection unit tests exercise precedence helpers. The local TCP and Artery runs here explicitly selected GraphStage.In 2.1.0 or 2.2.0, reassess whether downstream validation and binary compatibility permit retiring
TlsModule, its materializer island/phase, andTLSActor, then pruning the unused buffering/Pump/Transfer chain. Until then, keep them available. The still-usedTlsUtils.applySessionParametersis shared with the GraphStage; any file move must preserve its binary name and method shape.Do not blanket-delete
ActorSubscriberMessageorActorPublisher.scala: current GraphStage internals use the former; the latter owns shutdown sentinels used by the GraphInterpreter, fanout bridge, subscriber management, and hubs. Actor-ref sources, stream refs, TCP and file/stream I/O adapters are already GraphStages. Interpreter/supervisor actors are runtime infrastructure, outside this operator migration.Validation
git diff --checkpassed. The localsortImportsattempt hit a scalameta linkage error; unrelated edits were removed.These fixes, the default-engine switch, normal buffer-memory policy, and eventual legacy-class retirement remain separate changes with their own acceptance evidence.
Fix TLS GraphStage buffer allocation and growth
Preserve accepted TLS plaintext on output cancellation
Fix TLS GraphStage buffer allocation and growth
Preserve accepted TLS plaintext on output cancellation
fix: address TLS buffer sizing review feedback apache#3728
fix: address TLS cancellation review feedback apache#3727
docs: clarify TLS cancellation and peer completion ordering apache#3727
He-Pin commentedon Oct 29, 2026
Motivation
This corrects the TcpSpec conclusions in the comment created on 2026-10-28 at 22:20:27Z, without rewriting that comment.
Modification
hell/hellofailure does not show that TLS had accepted the finalobefore observing plaintext-output cancellation. In every failing diagnostic run, the server-side event order wasaccept(h), accept(e), accept(l), accept(l), cancel, accept(o).TcpSpec.scala, its four-argument call, closing policy, andhelloassertion are unchanged in the PR. The referenced31f6681c00is superseded by205155160a.h/e/l/lhad been accepted before cancellation. A lateonPush(o)was observed afterwards on both revisions; saying thatonever entered the operator would be inaccurate.Result
#3727 fixes the separate loss of plaintext already accepted at cancellation; it does not eliminate the original TcpSpec failure. The observed echo failure is a cancellation-versus-delivery race, outside that drain guarantee. These logs establish the event order at TLS, not the exact upstream buffer responsible. The follow-up is #3729.
The drain also applies when plaintext-output cancellation is observed before peer transport-input completion. If transport-input completion is observed first, the existing inbound-close contract discards accepted plaintext, matching the legacy actor engine.
Tests
On 2026-10-29, using OpenJDK 25.0.2, macOS arm64, Scala 2.13.18, and the unchanged TLS 1.3 TcpSpec case:
a55a008fdf(origin/main)a55a008fdf205155160a(#3727)205155160aAll failures returned
hell, expectedhello. Each repeat ran the original ScalaTest case in a fresh JVM using sbt-exported test classpath and JVM options;stream-tests/Test/fork = truewas confirmed, and GraphStage was explicitly selected by system property.A separate diagnostic batch ran 20 times per engine per revision. GraphStage failed 11/20 (55%) on main and 12/20 (60%) on the PR, always with the event order above. Successful GraphStage runs accepted
hellobefore cancellation. Both legacy batches passed 20/20 and loggedhelloat receipt andChoppingBlock.chopInto. Diagnostic changes were local only and removed.These are shared-workstation observations, not comparable failure-probability estimates: the one-minute load ranged from 5.76–6.72 during the main baseline and 19.40–34.44 during the PR baseline; logging can also affect scheduling. Earlier individual passes do not establish that this intermittent case was fixed.
References
Refs #2860, #3727, and #3729. Buffer allocation/growth remains separate in #3728.
Preserve TLS plaintext ordering and drain on cancellation apache#3729
Preserve TLS plaintext ordering and drain on cancellation apache#3729
Preserve TLS plaintext ordering and drain on cancellation #3729 (#3727)
Fix TLS GraphStage buffer allocation and growth apache#3728
Fix TLS GraphStage buffer allocation and growth apache#3728
Fix TLS GraphStage buffer allocation and growth #3728 (#3728)