Namespaces

Classes

1: <?php
2:
3: declare(strict_types=1);
4:
5: /**
6: * This file is part of the Nexus MCP SDK package.
7: *
8: * (c) 2026 John Paul E. Balandan, CPA <paulbalandan@gmail.com>
9: *
10: * For the full copyright and license information, please view
11: * the LICENSE file that was distributed with this source code.
12: */
13:
14: namespace Nexus\Mcp\Client\Transport;
15:
16: use Nexus\Assert\Assert;
17: use Nexus\Mcp\Core\Exception\SupervisionExhaustedException;
18: use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
19: use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
20: use Nexus\Mcp\Core\Exception\TransportNotStartedException;
21: use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
22: use Nexus\Mcp\Core\Transport\ReconnectingTransportInterface;
23: use Nexus\Mcp\Core\Transport\SendContext;
24: use Nexus\Mcp\Core\Transport\Subscription;
25: use Nexus\Mcp\Core\Transport\SubscriptionInterface;
26: use Nexus\Mcp\Core\Transport\SupervisableTransportInterface;
27: use Nexus\Mcp\Core\Transport\TransportEvents;
28: use Nexus\Mcp\Core\Transport\TransportState;
29: use Psr\Log\LoggerInterface;
30: use Psr\Log\NullLogger;
31: use Revolt\EventLoop;
32:
33: /**
34: * Transport that mints each connection from a factory and respawns the peer when one ends without
35: * `close()` having been called. Forwards the caller's listeners to whichever connection is current.
36: *
37: * @see docs/transports.md for the close, budget and retry semantics.
38: */
39: final class SupervisedTransport implements ReconnectingTransportInterface
40: {
41: /**
42: * Seconds the restart count is measured over before it starts again from zero.
43: */
44: public const float DEFAULT_RESTART_WINDOW = 60.0;
45:
46: private const string LABEL = 'Supervised client';
47:
48: private readonly TransportEvents $events;
49: private readonly LoggerInterface $logger;
50:
51: /**
52: * @var \Closure(): float
53: */
54: private readonly \Closure $clock;
55:
56: private TransportState $state = TransportState::Idle;
57: private ?SupervisableTransportInterface $inner = null;
58:
59: /**
60: * False exactly while a live connection still owes the caller a close.
61: */
62: private bool $connectionEnded = true;
63:
64: /**
65: * Respawns counted so far in the current window.
66: */
67: private int $restarts = 0;
68:
69: /**
70: * When the current window opened, as a reading from `$clock`. Zero until the first respawn.
71: */
72: private float $windowStartedAt = 0.0;
73:
74: /**
75: * @var list<SubscriptionInterface>
76: */
77: private array $subscriptions = [];
78:
79: /**
80: * @var array<int, \Closure(): void>
81: */
82: private array $reconnectListeners = [];
83:
84: private ?string $respawnWatcher = null;
85:
86: /**
87: * True from the moment a replacement is decided on until it is serving. Tracked apart from the
88: * watcher because arming one happens after a retire that suspends.
89: */
90: private bool $respawning = false;
91:
92: /**
93: * @param \Closure(): SupervisableTransportInterface $factory Mints one connection, called once per spawn.
94: * @param int $maxRestarts Respawns allowed within one window before giving up.
95: * @param float $restartDelay Seconds to wait before each respawn.
96: * @param float $restartWindow Seconds the restart count is measured over.
97: * @param null|\Closure(): float $clock Reads the current time in **seconds**, replaceable so the window boundary is exact under test. A source in other units silently makes the budget unspendable.
98: */
99: public function __construct(
100: private readonly \Closure $factory,
101: private readonly int $maxRestarts = 3,
102: private readonly float $restartDelay = 0.1,
103: LoggerInterface $logger = new NullLogger(),
104: private readonly float $restartWindow = self::DEFAULT_RESTART_WINDOW,
105: ?\Closure $clock = null,
106: ) {
107: Assert::that($maxRestarts)->isPositiveInt('maxRestarts must be a positive integer, {value} given.');
108: Assert::that($restartDelay)->isBetween(0.0, \PHP_FLOAT_MAX, message: 'restartDelay must not be negative, {value} given.');
109: Assert::that($restartWindow)->isBetween(\PHP_FLOAT_EPSILON, \PHP_FLOAT_MAX, message: 'restartWindow must be positive, {value} given.');
110:
111: $this->logger = $logger;
112: $this->clock = $clock ?? static fn(): float => microtime(true);
113: $this->events = new TransportEvents();
114: }
115:
116: #[\Override]
117: public function start(): void
118: {
119: match ($this->state) {
120: TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
121: TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
122: TransportState::Idle => null,
123: };
124:
125: // Only a connection that actually started makes this transport running, so a failed launch stays
126: // retryable instead of reporting itself as already started.
127: $this->spawn();
128: $this->state = TransportState::Running;
129: }
130:
131: #[\Override]
132: public function send(JsonRpcMessage $message, ?SendContext $context = null): void
133: {
134: if (TransportState::Idle === $this->state) {
135: throw new TransportNotStartedException(operation: 'send');
136: }
137:
138: // Released on close, and between a peer's death and its replacement being spawned. The connection
139: // the caller wrote against is gone, and the one that replaces it never carried the request.
140: if (null === $this->inner) {
141: throw new TransportAlreadyClosedException(operation: 'send');
142: }
143:
144: $this->inner->send($message, $context);
145: }
146:
147: #[\Override]
148: public function close(): void
149: {
150: // Every step below is idempotent, so a second call needs no guard of its own.
151: $this->state = TransportState::Closed;
152:
153: // The dead connection's own close already went out, and the caller has been holding its state for
154: // a replacement ever since. Abandoning that replacement owes them a second close, or they wait on
155: // a peer that is never coming. A replacement that is already up owes them nothing extra: the
156: // teardown below emits for it like any other live connection.
157: $abandonsAReplacement = $this->respawning && $this->connectionEnded;
158: $this->respawning = false;
159:
160: if (null !== $this->respawnWatcher) {
161: EventLoop::cancel($this->respawnWatcher);
162: $this->respawnWatcher = null;
163: }
164:
165: try {
166: // Told before delegating, so a peer that throws on the way down cannot swallow the signal.
167: $this->endConnection();
168:
169: if ($abandonsAReplacement) {
170: try {
171: $this->events->emitClose();
172: } catch (\Throwable $e) {
173: // Reached from a loop callback on the exhausted-budget path, and one listener failing
174: // must not strand the rest, which is the very thing this emission exists to prevent.
175: $this->events->emitError($e);
176: }
177: }
178: } finally {
179: $this->retireConnection();
180: }
181: }
182:
183: #[\Override]
184: public function onMessage(\Closure $listener): SubscriptionInterface
185: {
186: return $this->events->onMessage($listener);
187: }
188:
189: #[\Override]
190: public function onError(\Closure $listener): SubscriptionInterface
191: {
192: return $this->events->onError($listener);
193: }
194:
195: #[\Override]
196: public function onDrain(\Closure $listener): SubscriptionInterface
197: {
198: return $this->events->onDrain($listener);
199: }
200:
201: #[\Override]
202: public function onClose(\Closure $listener): SubscriptionInterface
203: {
204: return $this->events->onClose($listener);
205: }
206:
207: #[\Override]
208: public function isReconnecting(): bool
209: {
210: return $this->respawning;
211: }
212:
213: #[\Override]
214: public function onReconnect(\Closure $listener): SubscriptionInterface
215: {
216: $id = spl_object_id($listener);
217: $this->reconnectListeners[$id] = $listener;
218:
219: return new Subscription(function () use ($id): void {
220: unset($this->reconnectListeners[$id]);
221: });
222: }
223:
224: /**
225: * Builds a connection, binds this instance's listeners to it, and starts it.
226: *
227: * Every path that supersedes a connection releases it first, so a peer that has been replaced holds
228: * none of these listeners and cannot emit through this instance afterwards.
229: */
230: private function spawn(): void
231: {
232: $inner = ($this->factory)();
233: $this->inner = $inner;
234:
235: // Set only once a connection exists, so a factory that throws leaves no close owed.
236: $this->connectionEnded = false;
237:
238: $this->subscriptions = [
239: $inner->onMessage($this->events->emitMessage(...)),
240: $inner->onError($this->events->emitError(...)),
241: $inner->onDrain($this->events->emitDrain(...)),
242: $inner->onClose($this->endConnection(...)),
243: $inner->onUnexpectedExit(function (?int $exitCode): void {
244: try {
245: $this->endConnection();
246: } finally {
247: // A close listener that throws aborts the chain, and must not take supervision with it.
248: $this->scheduleRespawn($exitCode);
249: }
250: }),
251: ];
252:
253: try {
254: $inner->start();
255: } catch (\Throwable $e) {
256: // A connection that never started is not one this instance can serve: it neither stays
257: // attached to the caller's listeners nor owes them a close.
258: $this->connectionEnded = true;
259: $this->releaseConnection();
260:
261: throw $e;
262: }
263: }
264:
265: /**
266: * Emits the caller's close for the current connection, once, whichever of the peer's death signals
267: * arrived first.
268: */
269: private function endConnection(): void
270: {
271: if ($this->connectionEnded) {
272: return;
273: }
274:
275: $this->connectionEnded = true;
276: $this->events->emitClose();
277: }
278:
279: private function scheduleRespawn(?int $exitCode): void
280: {
281: if (TransportState::Running !== $this->state) {
282: return;
283: }
284:
285: $now = ($this->clock)();
286:
287: // Counted over a moving window, so a peer that ran for a while and then died starts a fresh
288: // budget. Treating a served message as proof of health cannot work: the protocol layer replays
289: // its own state on every reconnect, so even a crash-looping peer answers something.
290: // The window opens at the first restart rather than at whatever the clock's origin happens to be:
291: // a monotonic source legitimately starts near zero, which would anchor it before the process ran.
292: if (0 === $this->restarts || $this->restartWindow < $now - $this->windowStartedAt) {
293: $this->restarts = 0;
294: $this->windowStartedAt = $now;
295: }
296:
297: ++$this->restarts;
298:
299: if ($this->restarts > $this->maxRestarts) {
300: $this->logger->error(
301: '{label} transport exhausted its restart budget of {budget}.',
302: ['label' => self::LABEL, 'budget' => $this->maxRestarts],
303: );
304:
305: try {
306: $this->events->emitError(new SupervisionExhaustedException($this->maxRestarts));
307: } finally {
308: $this->close();
309: }
310:
311: return;
312: }
313:
314: $this->logger->warning(
315: '{label} transport respawning the peer after an unexpected exit (code {exitCode}), attempt {attempt} of {budget}.',
316: ['label' => self::LABEL, 'exitCode' => $exitCode ?? 'unknown', 'attempt' => $this->restarts, 'budget' => $this->maxRestarts],
317: );
318:
319: // Set before retiring, which suspends on any transport that drains its streams on the way down.
320: // Anything the close already queued runs inside that suspension and must see a replacement coming.
321: $this->respawning = true;
322:
323: $this->retireConnection();
324:
325: // Retiring suspends on any transport that drains on the way down, so a close can land inside it.
326: // Arming now would resurrect a peer the caller has already been told is never coming, and nothing
327: // afterwards would ever close it.
328: if (TransportState::Running !== $this->state) {
329: return;
330: }
331:
332: // close() cancels this watcher, so reaching the callback proves supervision was still wanted.
333: $this->respawnWatcher = EventLoop::delay($this->restartDelay, function (): void {
334: $this->respawnWatcher = null;
335:
336: try {
337: $this->spawn();
338: } catch (\Throwable $e) {
339: // A replacement that cannot be built is itself a failed attempt, so it spends budget
340: // instead of escaping into the event loop's error handler.
341: $this->events->emitError($e);
342: $this->scheduleRespawn(null);
343:
344: return;
345: }
346:
347: // Starting suspends too, so the same close can land while the replacement comes up. It is
348: // nobody's peer then, and announcing it would have listeners write to a closed transport.
349: if (TransportState::Running !== $this->state) {
350: // The close that got here retired this connection on its way past: `spawn()` publishes
351: // it before `start()` suspends, so there is nothing left to release.
352: return;
353: }
354:
355: $this->respawning = false;
356:
357: // Announced only once the replacement is serving, so a listener that rebuilds per-connection
358: // state writes to a peer that can take it.
359: foreach ($this->reconnectListeners as $listener) {
360: try {
361: $listener();
362: } catch (\Throwable $e) {
363: // One listener's failure must not cost the rest theirs: the protocol layer rebuilding
364: // its state is in this chain, and losing it silently strands every open stream.
365: $this->events->emitError($e);
366: }
367: }
368: });
369: }
370:
371: /**
372: * Closes the current connection and drops this instance's listeners from it.
373: */
374: private function retireConnection(): void
375: {
376: $inner = $this->inner;
377:
378: try {
379: // Closing first keeps the peer's drain and close on the caller's chain, and forces EOF on a
380: // read loop the peer left parked on a stream something else still holds open.
381: $inner?->close();
382: } finally {
383: $this->releaseConnection();
384: }
385: }
386:
387: /**
388: * Drops this instance's listeners from the current connection and forgets it.
389: */
390: private function releaseConnection(): void
391: {
392: foreach ($this->subscriptions as $subscription) {
393: $subscription->dispose();
394: }
395:
396: $this->subscriptions = [];
397: $this->inner = null;
398: }
399: }
400:
Nexus MCP SDK API documentation generated by ApiGen dev-master