File Client/Transport/SupervisedTransport.php

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\Clock\HighResolutionStopwatch;
18:	use Nexus\Clock\Stopwatch;
19:	use Nexus\Mcp\Core\Exception\SupervisionExhaustedException;
20:	use Nexus\Mcp\Core\Exception\TransportAlreadyClosedException;
21:	use Nexus\Mcp\Core\Exception\TransportAlreadyStartedException;
22:	use Nexus\Mcp\Core\Exception\TransportNotStartedException;
23:	use Nexus\Mcp\Core\Schema\JsonRpc\JsonRpcMessage;
24:	use Nexus\Mcp\Core\Transport\ListenerHandle;
25:	use Nexus\Mcp\Core\Transport\ListenerHandleInterface;
26:	use Nexus\Mcp\Core\Transport\ReconnectingTransportInterface;
27:	use Nexus\Mcp\Core\Transport\SendContext;
28:	use Nexus\Mcp\Core\Transport\SupervisableTransportInterface;
29:	use Nexus\Mcp\Core\Transport\TransportEvents;
30:	use Nexus\Mcp\Core\Transport\TransportState;
31:	use Psr\Log\LoggerInterface;
32:	use Psr\Log\NullLogger;
33:	use Revolt\EventLoop;
34:
35:	/**
36:	* Transport that mints each connection from a factory and respawns the peer when one ends unasked.
37:	*
38:	* @see docs/transports/supervised.md for the close, budget and retry semantics.
39:	*/
40:	final class SupervisedTransport implements ReconnectingTransportInterface
41:	{
42:	/**
43:	* Seconds the restart count is measured over.
44:	*/
45:	public const float DEFAULT_RESTART_WINDOW = 60.0;
46:
47:	private readonly TransportEvents $events;
48:	private TransportState $state = TransportState::Idle;
49:	private ?SupervisableTransportInterface $inner = null;
50:	private bool $connectionEnded = true;
51:	private int $restartsInWindow = 0;
52:
53:	/**
54:	* A reading from `$stopwatch`, or zero until the first respawn.
55:	*/
56:	private float $windowStartedAt = 0.0;
57:
58:	/**
59:	* @var list<ListenerHandleInterface>
60:	*/
61:	private array $subscriptions = [];
62:
63:	/**
64:	* @var array<int, \Closure(): void>
65:	*/
66:	private array $reconnectListeners = [];
67:
68:	private ?string $respawnWatcher = null;
69:
70:	/**
71:	* True from the moment a replacement is decided on until it is serving.
72:	*/
73:	private bool $respawning = false;
74:
75:	/**
76:	* @param \Closure(): SupervisableTransportInterface $factory Mints one connection, called once per spawn.
77:	* @param int $maxRestarts Respawns allowed within one window before giving up.
78:	* @param float $restartDelay Seconds to wait before each respawn.
79:	* @param float $restartWindow Seconds the restart count is measured over.
80:	*/
81:	public function __construct(
82:	private readonly \Closure $factory,
83:	private readonly int $maxRestarts = 3,
84:	private readonly float $restartDelay = 0.1,
85:	private readonly LoggerInterface $logger = new NullLogger(),
86:	private readonly float $restartWindow = self::DEFAULT_RESTART_WINDOW,
87:	private readonly Stopwatch $stopwatch = new HighResolutionStopwatch(),
88:	) {
89:	Assert::that($maxRestarts)->isPositiveInt('maxRestarts must be a positive integer, {value} given.');
90:	Assert::that($restartDelay)->isBetween(0.0, \PHP_FLOAT_MAX, message: 'restartDelay must not be negative, {value} given.');
91:	Assert::that($restartWindow)->isBetween(\PHP_FLOAT_EPSILON, \PHP_FLOAT_MAX, message: 'restartWindow must be positive, {value} given.');
92:
93:	$this->events = TransportEvents::create($this->logger, 'Supervised client');
94:	}
95:
96:	#[\Override]
97:	public function start(): void
98:	{
99:	match ($this->state) {
100:	TransportState::Running => throw new TransportAlreadyStartedException(transport: self::class),
101:	TransportState::Closed => throw new TransportAlreadyClosedException(operation: 'start'),
102:	TransportState::Idle => null,
103:	};
104:
105:	$this->spawn();
106:	$this->state = TransportState::Running;
107:	}
108:
109:	#[\Override]
110:	public function send(JsonRpcMessage $message, ?SendContext $context = null): void
111:	{
112:	if (TransportState::Idle === $this->state) {
113:	throw new TransportNotStartedException(operation: 'send');
114:	}
115:
116:	if (null === $this->inner) {
117:	throw new TransportAlreadyClosedException(operation: 'send');
118:	}
119:
120:	$this->inner->send($message, $context);
121:	}
122:
123:	#[\Override]
124:	public function close(): void
125:	{
126:	$coldClose = TransportState::Idle === $this->state;
127:	$this->state = TransportState::Closed;
128:
129:	$abandonsAReplacement = $this->respawning && $this->connectionEnded;
130:	$this->respawning = false;
131:
132:	if (null !== $this->respawnWatcher) {
133:	EventLoop::cancel($this->respawnWatcher);
134:	$this->respawnWatcher = null;
135:	}
136:
137:	try {
138:	// The inner's own close relays drain first, then close, through the live subscriptions.
139:	$this->retireConnection();
140:	} finally {
141:	if ($coldClose) {
142:	$this->events->emitDrain();
143:	}
144:
145:	if ($abandonsAReplacement \|\| $coldClose) {
146:	try {
147:	$this->events->emitClose();
148:	} catch (\Throwable $e) {
149:	$this->events->emitError($e);
150:	}
151:	} else {
152:	$this->endConnection();
153:	}
154:	}
155:	}
156:
157:	#[\Override]
158:	public function onMessage(\Closure $listener): ListenerHandleInterface
159:	{
160:	return $this->events->onMessage($listener);
161:	}
162:
163:	#[\Override]
164:	public function onError(\Closure $listener): ListenerHandleInterface
165:	{
166:	return $this->events->onError($listener);
167:	}
168:
169:	#[\Override]
170:	public function onDrain(\Closure $listener): ListenerHandleInterface
171:	{
172:	return $this->events->onDrain($listener);
173:	}
174:
175:	#[\Override]
176:	public function onClose(\Closure $listener): ListenerHandleInterface
177:	{
178:	return $this->events->onClose($listener);
179:	}
180:
181:	#[\Override]
182:	public function isReconnecting(): bool
183:	{
184:	return $this->respawning;
185:	}
186:
187:	#[\Override]
188:	public function onReconnect(\Closure $listener): ListenerHandleInterface
189:	{
190:	$id = spl_object_id($listener);
191:	$this->reconnectListeners[$id] = $listener;
192:
193:	return new ListenerHandle(function () use ($id): void {
194:	unset($this->reconnectListeners[$id]);
195:	});
196:	}
197:
198:	private function spawn(): void
199:	{
200:	$inner = ($this->factory)();
201:	$this->inner = $inner;
202:
203:	$this->connectionEnded = false;
204:
205:	$this->subscriptions = [
206:	$inner->onMessage($this->events->emitMessage(...)),
207:	$inner->onError($this->events->emitError(...)),
208:	$inner->onDrain($this->events->emitDrain(...)),
209:	$inner->onClose($this->endConnection(...)),
210:	$inner->onUnexpectedExit(function (?int $exitCode): void {
211:	try {
212:	$this->endConnection();
213:	} finally {
214:	$this->scheduleRespawn($exitCode);
215:	}
216:	}),
217:	];
218:
219:	try {
220:	$inner->start();
221:	} catch (\Throwable $e) {
222:	$this->connectionEnded = true;
223:	$this->releaseConnection();
224:
225:	throw $e;
226:	}
227:	}
228:
229:	private function endConnection(): void
230:	{
231:	if ($this->connectionEnded) {
232:	return;
233:	}
234:
235:	$this->connectionEnded = true;
236:	$this->events->emitClose();
237:	}
238:
239:	private function scheduleRespawn(?int $exitCode): void
240:	{
241:	if (TransportState::Running !== $this->state) {
242:	return;
243:	}
244:
245:	$now = $this->stopwatch->read();
246:
247:	if (0 === $this->restartsInWindow \|\| $this->restartWindow < $now - $this->windowStartedAt) {
248:	$this->restartsInWindow = 0;
249:	$this->windowStartedAt = $now;
250:	}
251:
252:	++$this->restartsInWindow;
253:
254:	if ($this->restartsInWindow > $this->maxRestarts) {
255:	$this->logger->error(
256:	'Supervised client transport exhausted its restart budget of {budget}.',
257:	['budget' => $this->maxRestarts],
258:	);
259:
260:	try {
261:	$this->events->emitError(new SupervisionExhaustedException($this->maxRestarts));
262:	} finally {
263:	$this->close();
264:	}
265:
266:	return;
267:	}
268:
269:	$this->logger->warning(
270:	'Supervised client transport respawning the peer after an unexpected exit (code {exitCode}), attempt {attempt} of {budget}.',
271:	['exitCode' => $exitCode ?? 'unknown', 'attempt' => $this->restartsInWindow, 'budget' => $this->maxRestarts],
272:	);
273:
274:	$this->respawning = true;
275:
276:	$this->retireConnection();
277:
278:	if (TransportState::Running !== $this->state) {
279:	return;
280:	}
281:
282:	$this->respawnWatcher = EventLoop::delay($this->restartDelay, function (): void {
283:	$this->respawnWatcher = null;
284:
285:	try {
286:	$this->spawn();
287:	} catch (\Throwable $e) {
288:	$this->events->emitError($e);
289:	$this->scheduleRespawn(null);
290:
291:	return;
292:	}
293:
294:	if (TransportState::Running !== $this->state) {
295:	return;
296:	}
297:
298:	$this->respawning = false;
299:
300:	foreach ($this->reconnectListeners as $listener) {
301:	try {
302:	$listener();
303:	} catch (\Throwable $e) {
304:	$this->events->emitError($e);
305:	}
306:	}
307:	});
308:	}
309:
310:	private function retireConnection(): void
311:	{
312:	$inner = $this->inner;
313:
314:	try {
315:	$inner?->close();
316:	} finally {
317:	$this->releaseConnection();
318:	}
319:	}
320:
321:	private function releaseConnection(): void
322:	{
323:	foreach ($this->subscriptions as $subscription) {
324:	$subscription->dispose();
325:	}
326:
327:	$this->subscriptions = [];
328:	$this->inner = null;
329:	}
330:	}
331:

Namespaces

Classes