🪄 The Fiber Illusion: What RxJS Taught Us About Real Async Pipelines in PHP

on July 18, 2026

Most developers meet RxJS through fluent syntax:

source.pipe(map(...), mergeMap(...), retry(...));

Memorable—and the least interesting part when you are designing a PHP pipeline engine.

We used RxJS as an architectural mirror for Darkwood Flow. Teams building scrapers, importers, and media pipelines keep asking whether they need “reactive” APIs. Our question was sharper:

What can Flow learn from RxJS without becoming RxJS?

RxJS describes values over time under an explicit lifecycle: compose with pipe, start on subscribe, track active work, wait for completion, propagate errors, tear down on unsubscribe. Schedulers matter for timers; they are not what makes concurrent URL fetches work. The durable idea is the execution lifecycle.

Under the operator sugar, that machine looks like this:

subscription starts execution
→ values propagate through a chain
→ asynchronous inner work is tracked
→ completion waits for active work
→ errors terminate or restart the run
→ unsubscription triggers teardown

Flow already faces the same questions as it grows beyond simple stage composition:

• When does a pipeline start?
• When is it complete?
• What is still active?
• What happens on error?
• Who closes resources?
• How do we stop a run?

In Flow, each value travels in an instruction packet named Ip. Packets are admitted by strategies, scheduled by a Driver, and gathered after await(). We already supported Fibers, Amp, React, Swoole, and friends—and we already ran many packets through fetch → hash → save. What we lacked was honesty about native non-blocking I/O: overlapping real socket waits without requiring Amp or React as a hard dependency, and without pretending a Fiber around a blocking read is enough.

JavaScript gives RxJS an event loop. PHP CLI does not. A Fiber can suspend cooperative PHP execution; it cannot turn a blocking socket call into overlapping I/O by itself. Flow adapts with explicit primitives:

PHP streams → non-blocking mode → stream_select() → Fibers → Driver → Flow

Takeaway: Steal RxJS’s lifecycle discipline. Keep Flow’s vocabulary (jobs, packets, Drivers). This study was never a detour into RxPHP—the interesting residue is principles, not types.

To keep ourselves honest, we ran a concrete pipeline—not artificial timers alone:

URLs → fetch concurrently → checksum → save → per-URL result → summary

Real sockets expose blocking behavior. Concurrency caps either hold or they do not. Timeouts and cleanup become visible. One failed URL must not erase five successes when that is the product rule. Artificial delay() demos prove Fiber scheduling; they cannot prove open sockets share a stream_select wait.

Thesis: Flow should borrow RxJS’s discipline around execution, concurrency, completion, and teardown—while remaining a PHP-native engine built on jobs, packets, strategies, and Drivers.

Put differently: we thought a Fiber-based Driver already gave Flow asynchronous execution. Studying RxJS forced us to define what “async” means. The rest of this article is that discovery.

Under the operators: how RxJS actually runs

We inspected RxJS 8.0.0-alpha.14 (core types in @rxjs/observable). The package layout is version-specific; the lifecycle principles are broader than that alpha.

from(urls).pipe(
  mergeMap(fetchImage, 4),
  map(hashBinary),
  retry(2),
  finalize(cleanup),
);

Composition is not execution

Observable.pipe left-folds operators into new Observables. Nothing fetches yet. Operators in this tree typically return new Observable(subscriber => { ... }); the older lift path is deprecated toward v8. Flow already separates composition (FlowFactory / fn) from execution (await())—same split, different names.

Subscription starts the run

subscribe wraps the consumer in a Subscriber, connects the producer, and registers teardown. Operators subscribe upstream; next / error / complete push sinkward. Helpers such as operate({ destination, ... }) link children into the destination’s teardown tree, so unsubscribe cascades.

A Subscription stores finalizers and runs them idempotently on unsubscribe. Terminal signals stop notifications and unsubscribe. Cleanup is part of ending a run. Cold sources (from(array) included) isolate runs: each subscription gets its own producer.

mergeMap tracks active inners

mergeMap(project, concurrent) delegates to shared merge logic: an active-inner count, a buffer for overflow, and completion only when the outer source is done, the buffer is empty, and no inners remain. Inner inputs go through from(project(...)).

We did not need mergeMap in PHP. We needed its guarantee: bounded in-flight work and completion that waits for every active operation.

retry and finalize

On error, retry unsubscribes the current inner subscription and resubscribes the same source. For cold from(urls), that restarts the list—powerful, and often wrong for batch I/O. finalize(callback) registers cleanup on the outer subscription (complete, error, or unsubscribe), not between retry attempts.

What to steal, and what not to

This is the heart of the study.

Analogies help—and stop at analogy:

• Subscription ≈ a future Flow execution handle
• Scheduler ≈ Driver
• mergeMap(n) ≈ bounded concurrent stage (MaxIpStrategy + yielding jobs)

Takeaway: Copy the lifecycle contract. Do not copy the type system.

What Flow already had

Before the study, Flow was not a blank slate. That baseline is what made the Fiber illusion convincing.

We could already:

• Compose stages with FlowFactory (Generator do-notation) or fn()
• Move values in Ip packets
• Run jobs (JobInterface or closures)
• Cap admission with MaxIpStrategy($n)
• Schedule via Drivers (FiberDriver by default, plus Amp, React, Swoole, …)
• Barrier on await() (void—results need a collector job)
• Route thrown failures through optional errorJob

build pipeline → push Ip(s) → await()
→ Driver admits packets, runs jobs, forwards results
→ when nothing remains queued or active, await returns

Pushing a packet enqueues it on the first stage. Inside await(), the Driver asks each stage’s strategy which packets may run, starts job Fibers (or coroutines), then forwards success downstream or invokes errorJob. Finished packets free concurrency slots.

Mental model: packets enter stages, strategies limit how many run, the Driver owns the wait.

With FiberDriver, multiple packets in flight, delay(), and MaxIpStrategy(4), it looked like asynchronous execution. Amp timer demos overlapped for real. Native blocking stream I/O was the blind spot—and we were wrong about what “async” meant there.

The Fiber illusion

What we believed

Push four URLs → admit four → start four Fibers → await(). If jobs “did I/O,” the waits must overlap.

RxJS made that assumption uncomfortable: if mergeMap tracks active inners, what was Flow tracking—packets, Fibers, or sockets?

What broke the story

file_get_contents($url); // still blocks the process

Inside a FiberDriver job, that call never suspends on readiness. The Driver waits on the block. MaxIpStrategy(4) may admit four packets while the process still serializes at the socket—or freezes the scheduler while looking concurrent.

Same trap: blocking fread. Fibers help only when the job yields. Without a readiness API, there is nothing to yield on.

A Fiber can suspend PHP execution. It cannot make a blocking socket call non-blocking.

RxJS forced the vocabulary: admission limits are not I/O overlap. Flow had to adopt that honesty.

The stack we needed

non-blocking streams
→ stream_select()
→ Fiber suspension on wait
→ readiness-based resume
→ Driver await loop
→ Flow stages

That stack became the center of the study—and why StreamSelectDriver entered darkwood/flow.

Takeaway: “Async Driver” without a yield-on-readiness story is a slogan. With stream_select, it becomes a contract.

Concrete results: the image pipeline

The demo lives in content/rxjs-flow and consumes local Flow via a Composer path symlink. A short-lived demo-local engine proved stream_select, then died on purpose. One reusable runtime only.

php bin/console app:fetch-images --concurrency=4 --timeout=5

Concurrency sits on the fetch stage via MaxIpStrategy($concurrency). HTTP 404 is a soft per-item error on DTOs—not errorJob—so five successes coexist with one failure. Traces emit flow.started, item.started / completed / failed, flow.completed.

Representative run (concurrency=2):

[flow.started] items=6 concurrency=2
[item.started] url=http://127.0.0.1:8765/image-a.png
[item.started] url=http://127.0.0.1:8765/image-b.png
[item.started] url=http://127.0.0.1:8765/image-c.png
[item.completed] url=…/image-a.png checksum=… path=…
…
[item.failed] url=…/fail.png error=HTTP 404
[flow.completed] ok=5 failed=1 elapsed_ms=711.23

Fixtures use php -S (single-threaded) with short route delays so pool hand-off is visible. Client-side capped waits are real; wall-clock speedup needs a multi-connection server. The fetch Job is teaching HTTP—redirects, TLS edge cases, chunked quirks, proxies, compression, and partial I/O are out of scope. Production clients still belong in Jobs; Drivers stay generic.

Next acceptance test (expected, not claimed as measured here):

5 × ~1s delayed endpoints on a multi-connection backend
concurrency=1 → ~5s
concurrency=4 → ~2s

StreamSelectDriver: what we shipped

StreamSelectDriver multiplexes native stream readiness on Fibers:

$ready = $driver->waitReadable($stream, $timeoutSeconds);
$ready = $driver->waitWritable($stream, $timeoutSeconds);

Both return bool and must run inside a job Fiber. Internally the Driver:

1. Records Fiber, stream, mode, optional deadline
2. Suspends
3. Builds stream_select sets from pending waits
4. Resumes ready Fibers with true, timed-out waits with false
5. Still resumes plain delay() suspensions on the normal loop tick

Same DriverInterface as other Drivers (async, defer, await, delay, tick), plus these wait APIs Jobs opt into by type.

MaxIpStrategy FIFO fix: at capacity, do not pull and re-queue (that rotated order). Concurrent traces made the bug obvious.

Why not enhance FiberDriver? Default Fiber resume ≠ readiness-gated resume. A dedicated Driver keeps the scheduling rule explicit.

Driver     → schedule, wait, resume, deadlines
Job        → protocol + application work
IpStrategy → packet admission / concurrency
Collector  → terminal aggregate

$driver = new StreamSelectDriver();
$flow = (new FlowFactory($driver))->create(static function () use (...) {
    yield [$fetchJob, null, new MaxIpStrategy($concurrency)];
    yield $hashJob;
    yield $saveJob;
    yield $collectorFn;
});

Takeaway: Jobs yield. Drivers wait. HTTP never lives in the Driver—or every protocol invents a framework.

What “concurrency=4” actually limits

A concurrency limit is only meaningful if the runtime can say exactly what it limits.

MaxIpStrategy(4) caps packets processing at one stage. Alone, it does not create overlapping network waits.

With blocking I/O, the cap is cosmetic. With StreamSelectDriver and yielding fetch jobs:

MaxIpStrategy(4)
+ non-blocking fetch
+ StreamSelectDriver
→ ≤ 4 fetch packets admitted
→ ≤ 4 fetch Fibers active
→ ≤ 4 sockets waiting on select

That is Flow’s translation of mergeMap(..., 4). Scope is one stage: hash/save may still overlap the next fetch—that is pipeline overlap, not a broken cap.

Errors, completion, cleanup

Collect vs fail-fast. Batch imports want soft per-item errors; critical publishes may need errorJob or process failure. Flow supports both. Wait-layer timeouts already exist; full retry/timeout operators can wait until patterns repeat across apps.

Completion means: source exhausted, queues empty, no active jobs, no pending stream waits, resources closed. Today that barrier is await() plus a collector. A future $flow->start() / Execution (cancel, result, errors) is optional ergonomics—not required for this slice.

Cleanup is finally + wait deadlines—RxJS teardown in PHP clothes, not destructor hope. Structured traces beat an internal event bus for day-one observability.

Mapping RxJS → Flow

Keep: lazy compose / explicit run, multi-item pipelines, bounded concurrency, deterministic completion, cleanup, per-run isolation, observability.
Simplify: jobs + packets; Drivers; collectors—not a public next/error/complete protocol; limits—not operator names.
Postpone: public Execution, cancel tokens, first-class retry/timeout ops, switchMap / exhaustMap, hot streams.
Reject: RxJS-shaped public API, Subjects, Promise/operator zoos, HTTP-in-Driver, dual engines.

Flow stays a mature, pragmatic pipeline engine with a real async executor—not a reactive framework wearing PHP syntax.

Where Flow goes next

Keep:   lifecycle discipline (start → bound concurrency → wait → tear down)
Shape:  Job + Ip + MaxIpStrategy + Driver + await + collector
Next:   multi-connection overlap benchmarks
        cancel tokens on StreamSelectDriver waits
        richer await ergonomics if collectors become painful

Darkwood Flow is not becoming RxJS for PHP. It is doubling down on a PHP-native pipeline engine: bounded concurrency, readiness-based I/O, deterministic completion, and explicit cleanup—with Drivers that can prove their waits.

When a job waits on I/O, does the runtime actually wait with it—or does it only pretend?

Resources

• Demo repository: github.com/matyo91/rxjs-flow
• Presentation slides: github.com/matyo91/slidewire (/slides/the-fiber-illusion)