From b5937aa49775ccd91430456d6c71ed80e05f6896 Mon Sep 17 00:00:00 2001 From: Chad Sikorra Date: Sat, 18 Jul 2026 12:36:22 -0400 Subject: [PATCH 1/2] Update the the sync handler to use ResponseStream. Narrow the queue on the handler context to just the interface that exposes no queue writing, only some things that certain operations still need to do. --- .../Container/HandlerContainerProvider.php | 4 +- .../Ldap/Protocol/Factory/HandlerContext.php | 4 +- .../ServerSyncHandler.php | 80 +++++++++------ .../Ldap/Server/ConnectionHandlerBuilder.php | 2 +- .../Sync/Provider/SyncPersistStreamer.php | 97 +++++++++++-------- .../Factory/ProtocolHandlerProviderTest.php | 4 +- .../ServerSyncHandlerTest.php | 40 ++++---- .../Sync/Provider/SyncPersistStreamerTest.php | 56 +++++++---- 8 files changed, 170 insertions(+), 117 deletions(-) diff --git a/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php b/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php index 54449463..0669e6cf 100644 --- a/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php +++ b/src/FreeDSx/Ldap/Container/HandlerContainerProvider.php @@ -82,7 +82,7 @@ private function makeFactoryMap(Container $container): ProtocolHandlerFactoryMap HandlerId::StartTls->value => static fn(HandlerContext $context): ServerProtocolHandlerInterface => new ServerStartTlsHandler( options: $container->get(ServerOptions::class), - connection: $context->queue, + connection: $context->connection, eventLogger: $context->eventLogger, ), HandlerId::UnsupportedExtended->value => static fn(): ServerProtocolHandlerInterface @@ -178,7 +178,6 @@ private function makeSyncHandler( if ($journal !== null) { $stream = new ChangeStream($journal); $streamer = new SyncPersistStreamer( - queue: $context->queue, backend: $backend, projector: $projector, stream: $stream, @@ -191,7 +190,6 @@ private function makeSyncHandler( } return new ServerSyncHandler( - queue: $context->queue, backend: $backend, projector: $projector, limits: $searchLimits ?? $options->makeSearchLimits(), diff --git a/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php b/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php index d20a4e81..645e3901 100644 --- a/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php +++ b/src/FreeDSx/Ldap/Protocol/Factory/HandlerContext.php @@ -13,7 +13,7 @@ namespace FreeDSx\Ldap\Protocol\Factory; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\ConnectionControl; use FreeDSx\Ldap\Server\Logging\EventLogger; use FreeDSx\Ldap\Server\PasswordPolicy\PasswordPolicyContext; use FreeDSx\Ldap\Server\RequestHistory; @@ -27,7 +27,7 @@ final readonly class HandlerContext { public function __construct( - public ServerQueue $queue, + public ConnectionControl $connection, public EventLogger $eventLogger, public RequestHistory $requestHistory, public ?PasswordPolicyContext $passwordPolicyContext = null, diff --git a/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php b/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php index d562e1ae..e3dac84b 100644 --- a/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php +++ b/src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerSyncHandler.php @@ -26,8 +26,9 @@ use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; +use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation; +use FreeDSx\Ldap\Protocol\Queue\Response\QueueWriterConfig; use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; use FreeDSx\Ldap\Server\Backend\LdapBackendInterface; use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream; use FreeDSx\Ldap\Server\Operation\SearchOperationResult; @@ -50,7 +51,6 @@ final class ServerSyncHandler implements ServerProtocolHandlerInterface use ServerSearchTrait; public function __construct( - private readonly ServerQueue $queue, private readonly LdapBackendInterface $backend, private readonly SyncResultProjector $projector, private readonly SearchLimits $limits = new SearchLimits(), @@ -60,8 +60,6 @@ public function __construct( ) {} /** - * Phase 1: sync (refresh + persist) stays a direct writer; it is folded into the writer in Phase 2. - * * @throws OperationException */ public function handleRequest( @@ -84,6 +82,7 @@ public function handleRequest( $this->assertModeSupported($mode); $baseDn = $this->assertBaseDnProvided($request); + $messageId = $message->getMessageId(); $latestSeq = $stream->latestSeq(); $sinceSeq = $this->resolveSince($control, $stream, $latestSeq); $state = new SearchResultState(); @@ -93,37 +92,62 @@ public function handleRequest( : $this->incrementalEntries($message, $request, $token, $streamer, $sinceSeq, $state); $cookie = (new SyncCookie($stream->origin(), $latestSeq))->encode(); + $outcome = fn(): SearchOperationResult => SearchOperationResult::success( + $message, + $state->entriesReturned, + ); if ($mode === SyncRequestControl::MODE_REFRESH_AND_PERSIST) { - $this->queue->sendMessages($this->withRefreshDone( - $entries, - $message->getMessageId(), - $sinceSeq === null - ? new SyncRefreshPresent(true, $cookie) - : new SyncRefreshDelete(true, $cookie), - )); - $streamer->persist( - $latestSeq, - $request, - $token, - $message->getMessageId(), - $baseDn, + $cancellation = new Cancellation(); + $boundary = $sinceSeq === null + ? new SyncRefreshPresent(true, $cookie) + : new SyncRefreshDelete(true, $cookie); + + return ResponseStream::streaming( + $this->concat( + $this->withRefreshDone( + $entries, + $messageId, + $boundary, + ), + $streamer->stream( + $latestSeq, + $request, + $token, + $messageId, + $baseDn, + $cancellation, + ), + ), + $outcome, + // Each change and keepalive flushes immediately for liveness; poll cancel after each. + new QueueWriterConfig(flushPerMessage: true, signalInterval: 1), + $cancellation, ); - } else { - // refreshDeletes: a delete phase (true) only for an incremental sync that sent explicit - // deletes; a full refresh is a present phase (false) the consumer reconciles by absence. - $this->queue->sendMessages($this->withSyncDone( + } + + // refreshDeletes: a delete phase (true) only for an incremental sync that sent explicit + // deletes; a full refresh is a present phase (false) the consumer reconciles by absence. + return ResponseStream::streaming( + $this->withSyncDone( $entries, - $message->getMessageId(), + $messageId, $baseDn, new SyncDoneControl($cookie, $sinceSeq !== null), - )); - } + ), + $outcome, + ); + } - return ResponseStream::none(SearchOperationResult::success( - $message, - $state->entriesReturned, - )); + /** + * @param Generator ...$streams + * @return Generator + */ + private function concat(Generator ...$streams): Generator + { + foreach ($streams as $stream) { + yield from $stream; + } } /** diff --git a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php index d868a957..a46a2529 100644 --- a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php +++ b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php @@ -352,7 +352,7 @@ private function makeProtocolHandlerProvider( routeResolver: $this->container->get(ServerProtocolHandlerFactory::class), factories: $this->container->get(ProtocolHandlerFactoryMap::class), context: new HandlerContext( - queue: $queue, + connection: $queue, eventLogger: $eventLogger, requestHistory: $requestHistory, passwordPolicyContext: $policyContext, diff --git a/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php b/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php index 466a64ab..fb6b9f0c 100644 --- a/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php +++ b/src/FreeDSx/Ldap/Sync/Provider/SyncPersistStreamer.php @@ -16,15 +16,13 @@ use FreeDSx\Ldap\Control\Sync\SyncDoneControl; use FreeDSx\Ldap\Entry\Dn; use FreeDSx\Ldap\Operation\LdapResult; -use FreeDSx\Ldap\Operation\Request\CancelRequest; use FreeDSx\Ldap\Operation\Request\SearchRequest; use FreeDSx\Ldap\Operation\Response\ExtendedResponse; use FreeDSx\Ldap\Operation\Response\SearchResultDone; use FreeDSx\Ldap\Operation\Response\SyncInfo\SyncNewCookie; use FreeDSx\Ldap\Operation\ResultCode; -use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation; use FreeDSx\Ldap\Server\Backend\LdapBackendInterface; use FreeDSx\Ldap\Server\Backend\Storage\Journal\Change\ChangeType; use FreeDSx\Ldap\Server\Backend\Storage\Journal\Change\PendingChange; @@ -33,9 +31,10 @@ use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream; use FreeDSx\Ldap\Server\Clock\Sleeper\SleeperInterface; use FreeDSx\Ldap\Server\Token\TokenInterface; +use Generator; /** - * Tails the change journal for a refreshAndPersist consumer, pushing changes until it cancels or disconnects. + * Tails the change journal for a refreshAndPersist consumer, yielding changes until it cancels or disconnects. * * @author Chad Sikorra */ @@ -47,7 +46,6 @@ public const DEFAULT_POLL_INTERVAL = 1.0; public function __construct( - private ServerQueue $queue, private LdapBackendInterface $backend, private SyncResultProjector $projector, private ChangeStream $stream, @@ -84,15 +82,20 @@ public function projectSince( } /** - * Runs the persist phase: pushes each change as it lands, advancing the cookie, until cancel or disconnect. + * The persist phase: yields each change as it lands, advancing the cookie, until cancel or disconnect. + * + * The writer polls the queue and offers any abandon/cancel into the shared token, which this loop reads. + * + * @return Generator */ - public function persist( + public function stream( int $startSeq, SearchRequest $request, TokenInterface $token, int $messageId, Dn $baseDn, - ): void { + Cancellation $cancellation, + ): Generator { $lastSeq = $startSeq; $origin = $this->stream->origin(); @@ -101,26 +104,35 @@ public function persist( // A trim past the consumer's position would silently drop changes: force a full re-sync instead. if ($latestSeq > $lastSeq && !$this->stream->retainsSince($lastSeq)) { - $this->sendRefreshRequired($messageId, $baseDn, $origin, $latestSeq); + yield $this->refreshRequired($messageId, $baseDn, $origin, $latestSeq); return; } if ($latestSeq > $lastSeq) { - $this->pushChanges($lastSeq, $request, $token, $messageId); + yield from $this->pushChanges($lastSeq, $request, $token, $messageId); $lastSeq = $latestSeq; } // Advances the client cookie, and doubles as a keepalive that surfaces a dead peer on the next poll. - $this->queue->sendMessage(new LdapMessageResponse( + yield new LdapMessageResponse( $messageId, new SyncNewCookie((new SyncCookie($origin, $lastSeq))->encode()), - )); + ); - $signal = $this->queue->peekForCancelSignal($messageId); + $signal = $cancellation->signal(); if ($signal !== null) { - $this->sendTerminal($signal, $messageId, $baseDn, $origin, $lastSeq); + // Abandon carries no response (RFC 4511 §4.11); only a Cancel gets an acknowledged close. + if ($cancellation->isCanceled()) { + yield from $this->terminal( + $signal->getMessageId(), + $messageId, + $baseDn, + $origin, + $lastSeq, + ); + } return; } @@ -129,18 +141,21 @@ public function persist( } } + /** + * @return Generator + */ private function pushChanges( int $sinceSeq, SearchRequest $request, TokenInterface $token, int $messageId, - ): void { + ): Generator { foreach ($this->projectSince($sinceSeq, $request, $token) as $result) { - $this->queue->sendMessage(new LdapMessageResponse( + yield new LdapMessageResponse( $messageId, $result->entry, $result->control, - )); + ); } } @@ -173,44 +188,40 @@ private function scopeFor(SearchRequest $request): ChangeScope }; } - private function sendTerminal( - LdapMessageRequest $signal, + /** + * @return Generator + */ + private function terminal( + int $cancelMessageId, int $messageId, Dn $baseDn, ReplicaId $origin, int $lastSeq, - ): void { - // Abandon carries no response (RFC 4511 §4.11); only a Cancel gets an acknowledged close. - if (!($signal->getRequest() instanceof CancelRequest)) { - return; - } - - $this->queue->sendMessage( - new LdapMessageResponse( - $messageId, - new SearchResultDone( - ResultCode::CANCELED, - $baseDn->toString(), - ), - new SyncDoneControl( - (new SyncCookie($origin, $lastSeq))->encode(), - true, - ), + ): Generator { + yield new LdapMessageResponse( + $messageId, + new SearchResultDone( + ResultCode::CANCELED, + $baseDn->toString(), ), - new LdapMessageResponse( - $signal->getMessageId(), - new ExtendedResponse(new LdapResult(ResultCode::SUCCESS)), + new SyncDoneControl( + (new SyncCookie($origin, $lastSeq))->encode(), + true, ), ); + yield new LdapMessageResponse( + $cancelMessageId, + new ExtendedResponse(new LdapResult(ResultCode::SUCCESS)), + ); } - private function sendRefreshRequired( + private function refreshRequired( int $messageId, Dn $baseDn, ReplicaId $origin, int $latestSeq, - ): void { - $this->queue->sendMessage(new LdapMessageResponse( + ): LdapMessageResponse { + return new LdapMessageResponse( $messageId, new SearchResultDone( ResultCode::SYNCHRONIZATION_REFRESH_REQUIRED, @@ -220,6 +231,6 @@ private function sendRefreshRequired( (new SyncCookie($origin, $latestSeq))->encode(), true, ), - )); + ); } } diff --git a/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php b/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php index 7865d5fc..1bed9ab6 100644 --- a/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php +++ b/tests/unit/Protocol/Factory/ProtocolHandlerProviderTest.php @@ -25,7 +25,7 @@ use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerFactoryMap; use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerProvider; use FreeDSx\Ldap\Protocol\Factory\ServerProtocolHandlerFactory; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\ConnectionControl; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerDispatchHandler; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerPagingHandler; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerPasswordModifyHandler; @@ -252,7 +252,7 @@ public function test_it_throws_when_password_policy_is_enabled_but_the_backend_i private function handlerContext(?PasswordPolicyContext $passwordPolicyContext = null): HandlerContext { return new HandlerContext( - queue: $this->createMock(ServerQueue::class), + connection: $this->createMock(ConnectionControl::class), eventLogger: new EventLogger(null), requestHistory: new RequestHistory(), passwordPolicyContext: $passwordPolicyContext, diff --git a/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php b/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php index 70d62a99..a032d94b 100644 --- a/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php +++ b/tests/unit/Protocol/ServerProtocolHandler/ServerSyncHandlerTest.php @@ -29,6 +29,7 @@ use FreeDSx\Ldap\Protocol\Authorization\AuthzId; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; use FreeDSx\Ldap\Protocol\Queue\ServerQueue; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerSyncHandler; use FreeDSx\Ldap\Search\Filters; @@ -148,15 +149,6 @@ protected function setUp(): void return $this->queue; }); $this->queue - ->method('sendMessage') - ->willReturnCallback(function (LdapMessageResponse ...$responses): ServerQueue { - foreach ($responses as $response) { - $this->sent[] = $response; - } - - return $this->queue; - }); - $this->queue ->method('peekForCancelSignal') ->willReturnCallback(fn(): ?LdapMessageRequest => array_shift($this->cancelSignals)); @@ -166,7 +158,6 @@ protected function setUp(): void filterEvaluator: $this->filterEvaluator, ); $this->persistStreamer = new SyncPersistStreamer( - queue: $this->queue, backend: $this->backend, projector: $this->syncProjector, stream: $this->changeStream, @@ -174,7 +165,6 @@ protected function setUp(): void ); $this->subject = new ServerSyncHandler( - queue: $this->queue, backend: $this->backend, projector: $this->syncProjector, changeStream: $this->changeStream, @@ -459,9 +449,9 @@ public function test_refresh_and_persist_full_refresh_ends_the_refresh_phase_wit // A pending cancel terminates the persist loop after the boundary is emitted. $this->cancelSignals = [new LdapMessageRequest(9, new CancelRequest(1))]; - $this->persistHandler()->handleRequest( + $this->drive( + $this->persistHandler(), $this->syncMessage(null, SyncRequestControl::MODE_REFRESH_AND_PERSIST), - $this->token, ); $boundary = null; @@ -483,7 +473,6 @@ public function test_refresh_and_persist_full_refresh_ends_the_refresh_phase_wit public function test_sync_is_declined_when_no_change_stream_is_available(): void { $handler = new ServerSyncHandler( - queue: $this->queue, backend: $this->backend, projector: new SyncResultProjector( accessControl: $this->accessControl, @@ -503,7 +492,6 @@ public function test_sync_is_declined_when_no_change_stream_is_available(): void private function persistHandler(): ServerSyncHandler { return new ServerSyncHandler( - queue: $this->queue, backend: $this->backend, projector: $this->syncProjector, changeStream: $this->changeStream, @@ -517,10 +505,26 @@ private function handle( int $mode = SyncRequestControl::MODE_REFRESH_ONLY, bool $reloadHint = false, ): OperationResult { - return $this->subject->handleRequest( + return $this->drive( + $this->subject, $this->syncMessage($cookie, $mode, $reloadHint), - $this->token, - )->outcome(); + ); + } + + /** + * Drives the handler's stream through the writer, which populates the queue and polls for cancellation. + */ + private function drive( + ServerSyncHandler $handler, + LdapMessageRequest $message, + ): OperationResult { + return (new ResponseWriter($this->queue))->write( + $handler->handleRequest( + $message, + $this->token, + ), + $message->getMessageId(), + ); } private function syncMessage( diff --git a/tests/unit/Sync/Provider/SyncPersistStreamerTest.php b/tests/unit/Sync/Provider/SyncPersistStreamerTest.php index 26b908dd..1e9e78b5 100644 --- a/tests/unit/Sync/Provider/SyncPersistStreamerTest.php +++ b/tests/unit/Sync/Provider/SyncPersistStreamerTest.php @@ -28,6 +28,10 @@ use FreeDSx\Ldap\Protocol\Authorization\AuthzId; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; +use FreeDSx\Ldap\Protocol\Queue\Response\Cancellation; +use FreeDSx\Ldap\Protocol\Queue\Response\QueueWriterConfig; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; use FreeDSx\Ldap\Protocol\Queue\ServerQueue; use FreeDSx\Ldap\Search\Filters; use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface; @@ -39,6 +43,7 @@ use FreeDSx\Ldap\Server\Backend\Storage\Journal\InMemoryChangeJournal; use FreeDSx\Ldap\Server\Backend\Storage\Journal\Read\ChangeStream; use FreeDSx\Ldap\Server\Backend\Storage\Journal\ReplicaId; +use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; use FreeDSx\Ldap\Server\Token\TokenInterface; use FreeDSx\Ldap\Sync\Provider\SyncPersistStreamer; use FreeDSx\Ldap\Sync\Provider\SyncResultProjector; @@ -89,9 +94,13 @@ protected function setUp(): void $this->queue = $this->createMock(ServerQueue::class); $this->queue - ->method('sendMessage') - ->willReturnCallback(function (LdapMessageResponse ...$responses): ServerQueue { + ->method('sendMessages') + ->willReturnCallback(function (iterable $responses): ServerQueue { foreach ($responses as $response) { + if (!$response instanceof LdapMessageResponse) { + continue; + } + $this->sent[] = $response; } @@ -116,7 +125,6 @@ protected function setUp(): void ->willReturn(true); $this->subject = new SyncPersistStreamer( - queue: $this->queue, backend: $this->backend, projector: new SyncResultProjector( accessControl: $accessControl, @@ -132,7 +140,7 @@ public function test_a_cancel_closes_the_stream_with_a_canceled_result_and_ackno { $this->cancelSignals = [new LdapMessageRequest(9, new CancelRequest(5))]; - $this->persist(); + $this->drive($this->subject); $done = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone); self::assertNotNull($done); @@ -156,7 +164,7 @@ public function test_an_abandon_closes_the_stream_without_any_response(): void { $this->cancelSignals = [new LdapMessageRequest(9, new AbandonRequest(5))]; - $this->persist(); + $this->drive($this->subject); self::assertNull( $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone), @@ -175,7 +183,7 @@ public function test_a_change_that_lands_during_the_loop_is_pushed_as_an_add(): $this->cancelSignals[] = new LdapMessageRequest(9, new CancelRequest(5)); }; - $this->persist(); + $this->drive($this->subject); $entry = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultEntry); self::assertNotNull($entry); @@ -207,7 +215,6 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require ->willReturn(new ReplicaId(self::ORIGIN)); $subject = new SyncPersistStreamer( - queue: $this->queue, backend: $this->backend, projector: new SyncResultProjector( accessControl: $this->createMock(AccessControlInterface::class), @@ -218,13 +225,7 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require pollInterval: 0.0, ); - $subject->persist( - 0, - $this->request(), - $this->token, - 5, - new Dn('dc=example,dc=com'), - ); + $this->drive($subject); $done = $this->firstSent(fn(LdapMessageResponse $m): bool => $m->getResponse() instanceof SearchResultDone); self::assertNotNull($done); @@ -236,14 +237,29 @@ public function test_a_trim_past_the_consumer_position_ends_with_refresh_require ); } - private function persist(): void + /** + * Drives the persist generator through the writer, which polls the queue and feeds the cancellation token. + */ + private function drive(SyncPersistStreamer $subject): void { - $this->subject->persist( - 0, - $this->request(), - $this->token, + $cancellation = new Cancellation(); + $stream = ResponseStream::streaming( + $subject->stream( + 0, + $this->request(), + $this->token, + 5, + new Dn('dc=example,dc=com'), + $cancellation, + ), + fn(): OperationOutcomeResult => OperationOutcomeResult::succeeded(), + new QueueWriterConfig(flushPerMessage: true, signalInterval: 1), + $cancellation, + ); + + (new ResponseWriter($this->queue))->write( + $stream, 5, - new Dn('dc=example,dc=com'), ); } From a821af3345ee9a49b7d4c5f6b6bafeb269d90068 Mon Sep 17 00:00:00 2001 From: Chad Sikorra Date: Sat, 18 Jul 2026 13:25:20 -0400 Subject: [PATCH 2/2] Make the middleware carry through the ResponseStream. This lets us use one middleware that handles all server queue writes (aside from auth, which is its own thing) --- .../Queue/Response/ResponseStream.php | 8 ++ .../Ldap/Server/ConnectionHandlerBuilder.php | 22 ++-- .../Server/Middleware/AssertionMiddleware.php | 4 +- .../AuthorizationResolutionMiddleware.php | 4 +- .../Ldap/Server/Middleware/BindMiddleware.php | 7 +- .../Middleware/CriticalControlMiddleware.php | 4 +- .../Server/Middleware/MetricsMiddleware.php | 10 +- .../Middleware/OperationAuditMiddleware.php | 66 ++---------- .../OperationAuthorizationMiddleware.php | 4 +- .../Server/Middleware/Pipeline/ChainStep.php | 4 +- .../Middleware/Pipeline/HandlerInvoker.php | 19 +--- .../Middleware/Pipeline/MiddlewareChain.php | 4 +- .../Pipeline/MiddlewareHandlerInterface.php | 4 +- .../Pipeline/MiddlewareInterface.php | 4 +- .../Server/Middleware/ReadOnlyMiddleware.php | 23 ++-- .../RequestValidationMiddleware.php | 4 +- .../Middleware/ResourceLimitMiddleware.php | 4 +- ...eware.php => ResponseWriterMiddleware.php} | 59 +++++++---- .../Operation/FailedOperationResult.php | 87 +++++++++++++++ .../Server/Proxy/ProxyRequestForwarder.php | 12 +-- .../Server/Proxy/ProxyRequestPipeline.php | 8 +- .../Middleware/CallbackMiddlewareHandler.php | 5 +- .../Middleware/RecordingMiddleware.php | 4 +- .../Middleware/RecordingMiddlewareHandler.php | 6 +- .../Middleware/StreamMiddlewareHandler.php | 33 ++++++ .../Middleware/StubMiddlewareHandler.php | 7 +- .../support/Middleware/ThrowingMiddleware.php | 4 +- .../Middleware/ThrowingMiddlewareHandler.php | 4 +- .../Server/Middleware/BindMiddlewareTest.php | 2 +- .../OperationAuditMiddlewareTest.php | 65 +++++++----- .../Middleware/ReadOnlyMiddlewareTest.php | 47 ++++---- ...t.php => ResponseWriterMiddlewareTest.php} | 100 +++++++++++++----- .../Proxy/ProxyRequestForwarderTest.php | 6 +- .../Server/Proxy/ProxyRequestPipelineTest.php | 2 +- 34 files changed, 388 insertions(+), 258 deletions(-) rename src/FreeDSx/Ldap/Server/Middleware/{OperationErrorMiddleware.php => ResponseWriterMiddleware.php} (51%) create mode 100644 src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php create mode 100644 tests/support/Middleware/StreamMiddlewareHandler.php rename tests/unit/Server/Middleware/{OperationErrorMiddlewareTest.php => ResponseWriterMiddlewareTest.php} (62%) diff --git a/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php b/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php index 61425b73..e3313ef7 100644 --- a/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php +++ b/src/FreeDSx/Ldap/Protocol/Queue/Response/ResponseStream.php @@ -146,4 +146,12 @@ public static function none(OperationResult $outcome): self $outcome, ); } + + /** + * An already-written response carrying only its resolved outcome, handed up past the writer. + */ + public static function resolved(OperationResult $outcome): self + { + return self::none($outcome); + } } diff --git a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php index a46a2529..1fc22778 100644 --- a/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php +++ b/src/FreeDSx/Ldap/Server/ConnectionHandlerBuilder.php @@ -56,7 +56,7 @@ use FreeDSx\Ldap\Server\Middleware\MetricsMiddleware; use FreeDSx\Ldap\Server\Middleware\OperationAuditMiddleware; use FreeDSx\Ldap\Server\Middleware\OperationAuthorizationMiddleware; -use FreeDSx\Ldap\Server\Middleware\OperationErrorMiddleware; +use FreeDSx\Ldap\Server\Middleware\ResponseWriterMiddleware; use FreeDSx\Ldap\Server\Middleware\Pipeline\HandlerInvoker; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareChain; use FreeDSx\Ldap\Server\Middleware\ReadOnlyMiddleware; @@ -393,28 +393,24 @@ private function makeRequestPipeline( ), // The token is resolved at this point, so per-identity limits can be attached. $this->container->get(ResourceLimitMiddleware::class), - new OperationErrorMiddleware( - $queue, + // Audits the resolved outcome; sits above the writer so a streamed result's count is final. + new OperationAuditMiddleware(new OperationAuditor($eventLogger)), + // The single sink: drains every operation response and renders any thrown failure. + new ResponseWriterMiddleware( + new ResponseWriter($queue), $backend, $options->getAccessControl(), ), - new OperationAuditMiddleware(new OperationAuditor($eventLogger)), - // Only present on a read-only replica; sits below OperationErrorMiddleware so a rejection renders, + // Only present on a read-only replica; sits below the writer so its referral is drained, // and before the ACL loop so a write short-circuits early. ...($replicaConfig !== null - ? [new ReadOnlyMiddleware( - $queue, - $replicaConfig, - )] + ? [new ReadOnlyMiddleware($replicaConfig)] : []), $this->container->get(CriticalControlMiddleware::class), $this->container->get(OperationAuthorizationMiddleware::class), $this->container->get(AssertionMiddleware::class), ], - new HandlerInvoker( - $handlerProvider, - new ResponseWriter($queue), - ), + new HandlerInvoker($handlerProvider), ); } } diff --git a/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php index 43abfd38..c08ee1ef 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/AssertionMiddleware.php @@ -17,12 +17,12 @@ use FreeDSx\Ldap\Control\PagingControl; use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Operation\Request\SearchRequest; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\AssertionEvaluator; use FreeDSx\Ldap\Server\AccessControl\OperationTargetDn; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** * Rejects an operation whose RFC 4528 assertion control does not match its target entry, before dispatch. @@ -40,7 +40,7 @@ public function __construct(private AssertionEvaluator $evaluator) {} public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $message = $context->message; $request = $message->getRequest(); diff --git a/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php index e04fe9af..1ca3e777 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/AuthorizationResolutionMiddleware.php @@ -17,10 +17,10 @@ use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\Authorization\DispatchAuthorizer; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use FreeDSx\Ldap\Server\PasswordPolicy\Decision\PasswordPolicyOutcome; use FreeDSx\Ldap\Server\PasswordPolicy\PasswordPolicyContext; @@ -45,7 +45,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $authorization = $this->dispatchAuthorizer->authorize($context->message); if ($authorization->requiresAuthentication()) { diff --git a/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php index 6e8a9e8a..4434b754 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/BindMiddleware.php @@ -16,12 +16,12 @@ use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\Authenticator; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Protocol\ServerAuthorization; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; use FreeDSx\Ldap\Server\Token\AnonToken; /** @@ -43,7 +43,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $request = $context->message->getRequest(); if (!$this->authorization->isAuthenticationRequest($request)) { @@ -60,8 +60,9 @@ public function process( ); } + // The authenticator writes the bind response itself (an interactive sub-protocol); carry only the outcome. $this->authorization->setToken($this->authenticator->bind($context->message)); - return OperationOutcomeResult::succeeded(); + return ResponseStream::resolved(OperationOutcomeResult::succeeded()); } } diff --git a/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php index 91a8e3af..da4027c7 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/CriticalControlMiddleware.php @@ -17,10 +17,10 @@ use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\Factory\HandlerRouteResolverInterface; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use function in_array; use function sprintf; @@ -44,7 +44,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $controls = $context->message->controls(); $routeId = $this->routeResolver->routeIdFor( $context->message->getRequest(), diff --git a/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php index 6b7716f3..a91a5591 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php @@ -21,6 +21,7 @@ use FreeDSx\Ldap\Operation\Request\SearchRequest; use FreeDSx\Ldap\Operation\Request\SimpleBindRequest; use FreeDSx\Ldap\Operation\ResultCode; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Metrics\MetricsRecorderInterface; use FreeDSx\Ldap\Server\Metrics\Observation\OperationObservation; use FreeDSx\Ldap\Server\Metrics\Rollup\OperationRollupCoordinator; @@ -28,7 +29,6 @@ use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationOutcome; -use FreeDSx\Ldap\Server\Operation\OperationResult; use Throwable; use function microtime; @@ -53,15 +53,16 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $request = $context->message->getRequest(); $operation = OperationType::classify($request); $this->recorder->operationStarted($operation); $startedAt = microtime(true); try { - $result = $next->handle($context); + $stream = $next->handle($context); } catch (OperationException $e) { + // Only front-of-chain failures (bind/authorization) reach here; operation failures resolve as outcomes. $this->record( $request, $operation, @@ -84,6 +85,7 @@ public function process( throw $e; } + $result = $stream->outcome(); $this->record( $request, $operation, @@ -92,7 +94,7 @@ public function process( $result->resultCode(), ); - return $result; + return $stream; } private function record( diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php index 2998010f..0bcd065e 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/OperationAuditMiddleware.php @@ -13,19 +13,15 @@ namespace FreeDSx\Ldap\Server\Middleware; -use FreeDSx\Ldap\Exception\OperationException; -use FreeDSx\Ldap\Exception\SchemaRuleException; -use FreeDSx\Ldap\Operation\Request\CompareRequest; -use FreeDSx\Ldap\Operation\Request\SearchRequest; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Logging\OperationAuditor; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\AuditableResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** - * Audits every operation outcome (the success result, or a thrown failure) the pipeline propagates back up. + * Audits every operation outcome the pipeline resolves (success or a writer-answered failure). * * @internal * @author Chad Sikorra @@ -34,23 +30,12 @@ { public function __construct(private OperationAuditor $auditor) {} - /** - * @throws OperationException - */ public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { - try { - $result = $next->handle($context); - } catch (OperationException $e) { - $this->recordFailure( - $context, - $e, - ); - - throw $e; - } + ): ResponseStream { + $stream = $next->handle($context); + $result = $stream->outcome(); if ($result instanceof AuditableResult) { $result->record( @@ -59,45 +44,6 @@ public function process( ); } - return $result; - } - - private function recordFailure( - ServerRequestContext $context, - OperationException $e, - ): void { - if ($e instanceof SchemaRuleException) { - $this->auditor->recordSchemaViolations( - $e->getViolations(), - $context->message, - $context->tokenOrFail(), - ); - } - - if ($context->message->getRequest() instanceof SearchRequest) { - $this->auditor->recordSearchFailure( - $context->message, - $e, - $context->tokenOrFail(), - ); - - return; - } - - if ($context->message->getRequest() instanceof CompareRequest) { - $this->auditor->recordCompareFailure( - $context->message, - $e, - $context->tokenOrFail(), - ); - - return; - } - - $this->auditor->recordFailure( - $context->message, - $e, - $context->tokenOrFail(), - ); + return $stream; } } diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php index 2ff8fb5c..0814aa1a 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/OperationAuthorizationMiddleware.php @@ -28,6 +28,7 @@ use FreeDSx\Ldap\Protocol\Factory\HandlerId; use FreeDSx\Ldap\Protocol\Factory\HandlerRouteResolverInterface; use FreeDSx\Ldap\Protocol\LdapMessageRequest; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerMonitorHandler; use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface; use FreeDSx\Ldap\Server\AccessControl\Rule\AttributeAccess; @@ -36,7 +37,6 @@ use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use FreeDSx\Ldap\Server\Token\TokenInterface; /** @@ -64,7 +64,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $this->authorizePrivilegedExtendedOperation($context); $routeId = $this->routeResolver->routeIdFor( diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php index 845b91c5..1f178ae1 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php +++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/ChainStep.php @@ -13,7 +13,7 @@ namespace FreeDSx\Ldap\Server\Middleware\Pipeline; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; /** * Binds a single middleware to the next handler in the chain. @@ -28,7 +28,7 @@ public function __construct( private MiddlewareHandlerInterface $next, ) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { return $this->middleware->process( $context, diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php index 58bc4700..825254f2 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php +++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/HandlerInvoker.php @@ -14,23 +14,19 @@ namespace FreeDSx\Ldap\Server\Middleware\Pipeline; use FreeDSx\Ldap\Protocol\Factory\ProtocolHandlerProviderInterface; -use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; /** - * Terminal handler that resolves the per-request handler, writes its response, and returns its outcome. + * Terminal handler that resolves and invokes the per-request protocol handler, returning its response stream. * * @internal * @author Chad Sikorra */ final readonly class HandlerInvoker implements MiddlewareHandlerInterface { - public function __construct( - private ProtocolHandlerProviderInterface $protocolHandlerProvider, - private ResponseWriter $responseWriter, - ) {} + public function __construct(private ProtocolHandlerProviderInterface $protocolHandlerProvider) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { $handler = $this->protocolHandlerProvider->get( $context->message->getRequest(), @@ -38,14 +34,9 @@ public function handle(ServerRequestContext $context): OperationResult $context->searchLimits(), ); - $stream = $handler->handleRequest( + return $handler->handleRequest( $context->message, $context->tokenOrFail(), ); - - return $this->responseWriter->write( - $stream, - $context->message->getMessageId(), - ); } } diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php index 5743543f..e0e05437 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php +++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareChain.php @@ -13,7 +13,7 @@ namespace FreeDSx\Ldap\Server\Middleware\Pipeline; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; /** * Nests an ordered list of middleware around a terminal handler. @@ -43,7 +43,7 @@ public function __construct( $this->pipeline = $pipeline; } - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { return $this->pipeline->handle($context); } diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php index fe3b9c70..58c7b130 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php +++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareHandlerInterface.php @@ -14,7 +14,7 @@ namespace FreeDSx\Ldap\Server\Middleware\Pipeline; use FreeDSx\Ldap\Exception\OperationException; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Socket\Exception\ConnectionException; /** @@ -29,5 +29,5 @@ interface MiddlewareHandlerInterface * @throws OperationException * @throws ConnectionException */ - public function handle(ServerRequestContext $context): OperationResult; + public function handle(ServerRequestContext $context): ResponseStream; } diff --git a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php index 953fc82e..e150aea8 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php +++ b/src/FreeDSx/Ldap/Server/Middleware/Pipeline/MiddlewareInterface.php @@ -14,7 +14,7 @@ namespace FreeDSx\Ldap\Server\Middleware\Pipeline; use FreeDSx\Ldap\Exception\OperationException; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Socket\Exception\ConnectionException; /** @@ -34,5 +34,5 @@ interface MiddlewareInterface public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult; + ): ResponseStream; } diff --git a/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php index 77953b07..1be26bfe 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/ReadOnlyMiddleware.php @@ -28,13 +28,12 @@ use FreeDSx\Ldap\Operation\Response\ModifyResponse; use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\LdapMessageResponse; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\ReplicaConfig; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; use function in_array; @@ -59,15 +58,12 @@ OperationType::PasswordModify, ]; - public function __construct( - private ServerQueue $queue, - private ReplicaConfig $replicaConfig, - ) {} + public function __construct(private ReplicaConfig $replicaConfig) {} public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $request = $context->message->getRequest(); $isWrite = in_array( @@ -87,12 +83,13 @@ public function process( ); } - $this->queue->sendMessage($this->referralResponse( - $request, - $context->message->getMessageId(), - )); - - return OperationOutcomeResult::failed(ResultCode::REFERRAL); + return ResponseStream::of( + [$this->referralResponse( + $request, + $context->message->getMessageId(), + )], + OperationOutcomeResult::failed(ResultCode::REFERRAL), + ); } private function referralResponse( diff --git a/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php index 1ea22fa3..7abe6dcb 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/RequestValidationMiddleware.php @@ -14,10 +14,10 @@ namespace FreeDSx\Ldap\Server\Middleware; use FreeDSx\Ldap\Exception\RequestValidationException; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use function in_array; use function sprintf; @@ -41,7 +41,7 @@ final class RequestValidationMiddleware implements MiddlewareInterface public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $messageId = $context->message->getMessageId(); if ($messageId === 0) { diff --git a/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php index 5bbfa508..1185aea4 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/ResourceLimitMiddleware.php @@ -13,7 +13,7 @@ namespace FreeDSx\Ldap\Server\Middleware; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; @@ -31,7 +31,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $limits = $this->searchLimitResolver->resolve($context->tokenOrFail()); return $next->handle($context->withSearchLimits($limits)); diff --git a/src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php b/src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php similarity index 51% rename from src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php rename to src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php index be637531..4884dc83 100644 --- a/src/FreeDSx/Ldap/Server/Middleware/OperationErrorMiddleware.php +++ b/src/FreeDSx/Ldap/Server/Middleware/ResponseWriterMiddleware.php @@ -13,59 +13,78 @@ namespace FreeDSx\Ldap\Server\Middleware; -use FreeDSx\Asn1\Exception\EncoderException; use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Protocol\Factory\ResponseFactory; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\MatchedDnAccessFilterTrait; use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface; use FreeDSx\Ldap\Server\Backend\LdapBackendInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; +use FreeDSx\Ldap\Server\Operation\FailedOperationResult; /** - * Translates an OperationException thrown anywhere below into the matching response. + * The single sink: drains the response stream to the queue, rendering any thrown OperationException as its response. + * + * A streaming handler runs during the drain, so mid-stream failures are answered here too (a partial stream is followed by its error terminal). * * @internal * @author Chad Sikorra */ -final readonly class OperationErrorMiddleware implements MiddlewareInterface +final readonly class ResponseWriterMiddleware implements MiddlewareInterface { use MatchedDnAccessFilterTrait; public function __construct( - private ServerQueue $queue, + private ResponseWriter $writer, private LdapBackendInterface $backend, private AccessControlInterface $accessControl, private ResponseFactory $responseFactory = new ResponseFactory(), ) {} - /** - * @throws EncoderException - */ public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { + $messageId = $context->message->getMessageId(); + try { - return $next->handle($context); + $outcome = $this->writer->write( + $next->handle($context), + $messageId, + ); } catch (OperationException $e) { - $this->queue->sendMessage($this->responseFactory->getStandardResponse( + $outcome = $this->writer->write( + $this->errorStream($context, $e), + $messageId, + ); + } + + return ResponseStream::resolved($outcome); + } + + private function errorStream( + ServerRequestContext $context, + OperationException $exception, + ): ResponseStream { + return ResponseStream::of( + [$this->responseFactory->getStandardResponse( $context->message, - $e->getCode(), - $e->getMessage(), + $exception->getCode(), + $exception->getMessage(), $this->filterMatchedDn( - $e->getMatchedDn(), + $exception->getMatchedDn(), $context->tokenOrFail(), $this->backend, $this->accessControl, ), - )); - - return OperationOutcomeResult::failed($e->getCode()); - } + )], + new FailedOperationResult( + $context->message, + $exception, + ), + ); } } diff --git a/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php b/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php new file mode 100644 index 00000000..dae8c107 --- /dev/null +++ b/src/FreeDSx/Ldap/Server/Operation/FailedOperationResult.php @@ -0,0 +1,87 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace FreeDSx\Ldap\Server\Operation; + +use FreeDSx\Ldap\Exception\OperationException; +use FreeDSx\Ldap\Exception\SchemaRuleException; +use FreeDSx\Ldap\Operation\Request\CompareRequest; +use FreeDSx\Ldap\Operation\Request\SearchRequest; +use FreeDSx\Ldap\Protocol\LdapMessageRequest; +use FreeDSx\Ldap\Server\Logging\OperationAuditor; +use FreeDSx\Ldap\Server\Token\TokenInterface; + +/** + * The outcome for an operation the response writer answered from a caught exception; audits by request type. + * + * @internal + * @author Chad Sikorra + */ +final readonly class FailedOperationResult implements AuditableResult +{ + public function __construct( + private LdapMessageRequest $message, + private OperationException $exception, + ) {} + + public function outcome(): OperationOutcome + { + return OperationOutcome::Failed; + } + + public function resultCode(): int + { + return $this->exception->getCode(); + } + + public function record( + OperationAuditor $auditor, + TokenInterface $token, + ): void { + if ($this->exception instanceof SchemaRuleException) { + $auditor->recordSchemaViolations( + $this->exception->getViolations(), + $this->message, + $token, + ); + } + + $request = $this->message->getRequest(); + + if ($request instanceof SearchRequest) { + $auditor->recordSearchFailure( + $this->message, + $this->exception, + $token, + ); + + return; + } + + if ($request instanceof CompareRequest) { + $auditor->recordCompareFailure( + $this->message, + $this->exception, + $token, + ); + + return; + } + + $auditor->recordFailure( + $this->message, + $this->exception, + $token, + ); + } +} diff --git a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php index 94701e76..3b718d23 100644 --- a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php +++ b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestForwarder.php @@ -28,11 +28,11 @@ use FreeDSx\Ldap\Protocol\Factory\ResponseFactory; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Protocol\Queue\ServerQueue; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** * Relays a request to the upstream connection and relays the response back to the client. @@ -50,7 +50,7 @@ public function __construct( /** * {@inheritDoc} */ - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { $message = $context->message; $request = $message->getRequest(); @@ -59,12 +59,12 @@ public function handle(ServerRequestContext $context): OperationResult $this->client->unbind(); $this->queue->close(); - return OperationOutcomeResult::succeeded(); + return ResponseStream::resolved(OperationOutcomeResult::succeeded()); } // Synchronous, sequential forwarding means nothing is ever in flight upstream to abandon. if ($request instanceof AbandonRequest) { - return OperationOutcomeResult::succeeded(); + return ResponseStream::resolved(OperationOutcomeResult::succeeded()); } try { @@ -80,7 +80,7 @@ public function handle(ServerRequestContext $context): OperationResult $e->getMatchedDn(), )); - return OperationOutcomeResult::failed($e->getCode()); + return ResponseStream::resolved(OperationOutcomeResult::failed($e->getCode())); } if ($request instanceof SearchRequest) { @@ -89,7 +89,7 @@ public function handle(ServerRequestContext $context): OperationResult $this->relaySingle($message, $response); } - return OperationOutcomeResult::succeeded(); + return ResponseStream::resolved(OperationOutcomeResult::succeeded()); } /** diff --git a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php index e3270c4f..61800224 100644 --- a/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php +++ b/src/FreeDSx/Ldap/Server/Proxy/ProxyRequestPipeline.php @@ -14,11 +14,11 @@ namespace FreeDSx\Ldap\Server\Proxy; use FreeDSx\Ldap\Operation\Request\ExtendedRequest; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; use FreeDSx\Ldap\Protocol\ServerProtocolHandler\ServerProtocolHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** * Proxy request pipeline: handles StartTLS locally and forwards everything else. @@ -36,7 +36,7 @@ public function __construct( /** * {@inheritDoc} */ - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { $request = $context->message->getRequest(); @@ -46,10 +46,10 @@ public function handle(ServerRequestContext $context): OperationResult $context->tokenOrFail(), ); - return $this->responseWriter->write( + return ResponseStream::resolved($this->responseWriter->write( $stream, $context->message->getMessageId(), - ); + )); } return $this->forwarder->handle($context); diff --git a/tests/support/Middleware/CallbackMiddlewareHandler.php b/tests/support/Middleware/CallbackMiddlewareHandler.php index 16b2a1e0..ba583c9e 100644 --- a/tests/support/Middleware/CallbackMiddlewareHandler.php +++ b/tests/support/Middleware/CallbackMiddlewareHandler.php @@ -14,6 +14,7 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; use Closure; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationResult; @@ -30,8 +31,8 @@ */ public function __construct(private Closure $callback) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { - return ($this->callback)($context); + return ResponseStream::resolved(($this->callback)($context)); } } diff --git a/tests/support/Middleware/RecordingMiddleware.php b/tests/support/Middleware/RecordingMiddleware.php index d80920d0..e3f6cfae 100644 --- a/tests/support/Middleware/RecordingMiddleware.php +++ b/tests/support/Middleware/RecordingMiddleware.php @@ -13,10 +13,10 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** * Records before/after markers around the next handler for ordering assertions. @@ -33,7 +33,7 @@ public function __construct( public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { $this->log->record('before:' . $this->label); $result = $next->handle($context); $this->log->record('after:' . $this->label); diff --git a/tests/support/Middleware/RecordingMiddlewareHandler.php b/tests/support/Middleware/RecordingMiddlewareHandler.php index 4d9688f0..e87152cf 100644 --- a/tests/support/Middleware/RecordingMiddlewareHandler.php +++ b/tests/support/Middleware/RecordingMiddlewareHandler.php @@ -13,10 +13,10 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; -use FreeDSx\Ldap\Server\Operation\OperationResult; /** * Terminal handler that records its invocation and captures the received context. @@ -32,11 +32,11 @@ public function __construct( private readonly string $label = 'terminal', ) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { $this->received = $context; $this->log->record($this->label); - return OperationOutcomeResult::succeeded(); + return ResponseStream::resolved(OperationOutcomeResult::succeeded()); } } diff --git a/tests/support/Middleware/StreamMiddlewareHandler.php b/tests/support/Middleware/StreamMiddlewareHandler.php new file mode 100644 index 00000000..e4c68266 --- /dev/null +++ b/tests/support/Middleware/StreamMiddlewareHandler.php @@ -0,0 +1,33 @@ + + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +namespace Tests\Support\FreeDSx\Ldap\Middleware; + +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; +use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; +use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; + +/** + * Terminal handler that returns a preconfigured response stream (e.g. a streaming one that fails mid-drain). + * + * @author Chad Sikorra + */ +final readonly class StreamMiddlewareHandler implements MiddlewareHandlerInterface +{ + public function __construct(private ResponseStream $stream) {} + + public function handle(ServerRequestContext $context): ResponseStream + { + return $this->stream; + } +} diff --git a/tests/support/Middleware/StubMiddlewareHandler.php b/tests/support/Middleware/StubMiddlewareHandler.php index 991a61b5..b9484991 100644 --- a/tests/support/Middleware/StubMiddlewareHandler.php +++ b/tests/support/Middleware/StubMiddlewareHandler.php @@ -13,12 +13,13 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Operation\OperationResult; /** - * Terminal handler that returns a preconfigured result. + * Terminal handler that resolves to a preconfigured outcome. * * @author Chad Sikorra */ @@ -26,8 +27,8 @@ { public function __construct(private OperationResult $result) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { - return $this->result; + return ResponseStream::resolved($this->result); } } diff --git a/tests/support/Middleware/ThrowingMiddleware.php b/tests/support/Middleware/ThrowingMiddleware.php index 29a5ff21..63fe33ed 100644 --- a/tests/support/Middleware/ThrowingMiddleware.php +++ b/tests/support/Middleware/ThrowingMiddleware.php @@ -13,10 +13,10 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use Throwable; /** @@ -31,7 +31,7 @@ public function __construct(private Throwable $throwable) {} public function process( ServerRequestContext $context, MiddlewareHandlerInterface $next, - ): OperationResult { + ): ResponseStream { throw $this->throwable; } } diff --git a/tests/support/Middleware/ThrowingMiddlewareHandler.php b/tests/support/Middleware/ThrowingMiddlewareHandler.php index 20245bcf..4669c311 100644 --- a/tests/support/Middleware/ThrowingMiddlewareHandler.php +++ b/tests/support/Middleware/ThrowingMiddlewareHandler.php @@ -13,9 +13,9 @@ namespace Tests\Support\FreeDSx\Ldap\Middleware; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; -use FreeDSx\Ldap\Server\Operation\OperationResult; use Throwable; /** @@ -27,7 +27,7 @@ { public function __construct(private Throwable $throwable) {} - public function handle(ServerRequestContext $context): OperationResult + public function handle(ServerRequestContext $context): ResponseStream { throw $this->throwable; } diff --git a/tests/unit/Server/Middleware/BindMiddlewareTest.php b/tests/unit/Server/Middleware/BindMiddlewareTest.php index 8fff69cb..a68f16eb 100644 --- a/tests/unit/Server/Middleware/BindMiddlewareTest.php +++ b/tests/unit/Server/Middleware/BindMiddlewareTest.php @@ -89,7 +89,7 @@ public function test_a_successful_bind_stores_the_token_and_short_circuits(): vo ); self::assertSame( OperationOutcome::Succeeded, - $result->outcome(), + $result->outcome()->outcome(), ); } diff --git a/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php b/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php index cce64f4e..41d32c81 100644 --- a/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php +++ b/tests/unit/Server/Middleware/OperationAuditMiddlewareTest.php @@ -29,6 +29,7 @@ use FreeDSx\Ldap\Server\Logging\OperationAuditor; use FreeDSx\Ldap\Server\Middleware\OperationAuditMiddleware; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; +use FreeDSx\Ldap\Server\Operation\FailedOperationResult; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; use FreeDSx\Ldap\Server\Operation\WriteOperationResult; use FreeDSx\Ldap\Server\Token\BindToken; @@ -89,7 +90,7 @@ public function test_it_does_not_record_a_non_auditable_result(): void ); } - public function test_it_returns_the_result_from_the_next_handler_unchanged(): void + public function test_it_returns_the_stream_from_the_next_handler_unchanged(): void { $result = OperationOutcomeResult::failed(); @@ -98,44 +99,48 @@ public function test_it_returns_the_result_from_the_next_handler_unchanged(): vo $this->subject->process( $this->context, new StubMiddlewareHandler($result), - ), + )->outcome(), ); } - public function test_it_audits_a_write_authorization_denial_and_rethrows(): void + public function test_it_audits_a_write_authorization_denial(): void { - $exception = new OperationException( - 'Denied.', - ResultCode::INSUFFICIENT_ACCESS_RIGHTS, + $this->auditFailure( + $this->context, + new OperationException( + 'Denied.', + ResultCode::INSUFFICIENT_ACCESS_RIGHTS, + ), ); - $this->processThrowing($this->context, $exception); - self::assertTrue($this->wasLogged('authz.denied.write')); } public function test_it_audits_a_search_authorization_denial(): void { $context = $this->contextFor((new SearchRequest(Filters::present('cn')))->base('dc=bar')); - $exception = new OperationException( - 'Denied.', - ResultCode::INSUFFICIENT_ACCESS_RIGHTS, - ); - $this->processThrowing($context, $exception); + $this->auditFailure( + $context, + new OperationException( + 'Denied.', + ResultCode::INSUFFICIENT_ACCESS_RIGHTS, + ), + ); self::assertTrue($this->wasLogged('authz.denied.read')); } public function test_it_audits_a_critical_control_rejection(): void { - $exception = new OperationException( - 'Critical control 1.2.3.4 is not supported.', - ResultCode::UNAVAILABLE_CRITICAL_EXTENSION, + $this->auditFailure( + $this->context, + new OperationException( + 'Critical control 1.2.3.4 is not supported.', + ResultCode::UNAVAILABLE_CRITICAL_EXTENSION, + ), ); - $this->processThrowing($this->context, $exception); - self::assertTrue($this->wasLogged('control.critical.rejected')); } @@ -157,12 +162,12 @@ public function test_it_records_schema_violations_from_a_schema_rule_exception() $violations, ); - $this->processThrowing($this->context, $exception); + $this->auditFailure($this->context, $exception); self::assertTrue($this->wasLogged('schema.violation')); } - public function test_it_rethrows_the_caught_operation_exception(): void + public function test_it_lets_an_exception_from_the_next_handler_propagate(): void { $exception = new OperationException('Boom.'); @@ -174,18 +179,20 @@ public function test_it_rethrows_the_caught_operation_exception(): void ); } - private function processThrowing( + /** + * A failure reaches the audit middleware as a resolved FailedOperationResult, not a thrown exception. + */ + private function auditFailure( ServerRequestContext $context, OperationException $exception, ): void { - try { - $this->subject->process( - $context, - new ThrowingMiddlewareHandler($exception), - ); - self::fail('Expected the OperationException to be re-thrown.'); - } catch (OperationException) { - } + $this->subject->process( + $context, + new StubMiddlewareHandler(new FailedOperationResult( + $context->message, + $exception, + )), + ); } private function wasLogged(string $event): bool diff --git a/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php b/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php index 85d9c300..3919dfa2 100644 --- a/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php +++ b/tests/unit/Server/Middleware/ReadOnlyMiddlewareTest.php @@ -27,12 +27,13 @@ use FreeDSx\Ldap\Operations; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; -use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; use FreeDSx\Ldap\ReplicaConfig; use FreeDSx\Ldap\Search\Filters; use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; use FreeDSx\Ldap\Server\Middleware\ReadOnlyMiddleware; +use FreeDSx\Ldap\Server\Operation\OperationOutcome; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\MockObject\MockObject; @@ -40,8 +41,6 @@ final class ReadOnlyMiddlewareTest extends TestCase { - private ServerQueue&MockObject $queue; - private MiddlewareHandlerInterface&MockObject $next; private ReplicaConfig $replicaConfig; @@ -50,17 +49,13 @@ final class ReadOnlyMiddlewareTest extends TestCase protected function setUp(): void { - $this->queue = $this->createMock(ServerQueue::class); $this->next = $this->createMock(MiddlewareHandlerInterface::class); $this->replicaConfig = new ReplicaConfig( (new ClientOptions()) ->setServers(['primary.example.com']) ->setPort(389), ); - $this->subject = new ReadOnlyMiddleware( - $this->queue, - $this->replicaConfig, - ); + $this->subject = new ReadOnlyMiddleware($this->replicaConfig); } /** @@ -104,31 +99,31 @@ public function test_a_write_is_referred_to_the_primary_by_default( ->expects(self::never()) ->method('handle'); - $sent = null; - $this->queue - ->expects(self::once()) - ->method('sendMessage') - ->willReturnCallback(function (LdapMessageResponse $message) use (&$sent): ServerQueue { - $sent = $message; - - return $this->queue; - }); - - $result = $this->subject->process( + $stream = $this->subject->process( $this->context($request), $this->next, ); self::assertSame( ResultCode::REFERRAL, - $result->resultCode(), + $stream->outcome()->resultCode(), + ); + self::assertSame( + OperationOutcome::Failed, + $stream->outcome()->outcome(), + ); + + $messages = [...$stream->messages]; + self::assertCount( + 1, + $messages, ); self::assertInstanceOf( LdapMessageResponse::class, - $sent, + $messages[0], ); - $response = $sent->getResponse(); + $response = $messages[0]->getResponse(); self::assertInstanceOf( $expectedResponse, $response, @@ -149,9 +144,6 @@ public function test_a_write_is_rejected_when_referral_is_disabled(): void $this->next ->expects(self::never()) ->method('handle'); - $this->queue - ->expects(self::never()) - ->method('sendMessage'); $this->expectException(OperationException::class); $this->expectExceptionCode(ResultCode::UNWILLING_TO_PERFORM); @@ -165,16 +157,13 @@ public function test_a_write_is_rejected_when_referral_is_disabled(): void public function test_a_read_passes_through_to_the_next_handler(): void { $context = $this->context(Operations::search(Filters::present('objectClass'))); - $expected = OperationOutcomeResult::succeeded(); + $expected = ResponseStream::none(OperationOutcomeResult::succeeded()); $this->next ->expects(self::once()) ->method('handle') ->with($context) ->willReturn($expected); - $this->queue - ->expects(self::never()) - ->method('sendMessage'); self::assertSame( $expected, diff --git a/tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php b/tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php similarity index 62% rename from tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php rename to tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php index 829f9f53..ea1d23ce 100644 --- a/tests/unit/Server/Middleware/OperationErrorMiddlewareTest.php +++ b/tests/unit/Server/Middleware/ResponseWriterMiddlewareTest.php @@ -18,24 +18,33 @@ use FreeDSx\Ldap\Exception\OperationException; use FreeDSx\Ldap\Operation\Request\DeleteRequest; use FreeDSx\Ldap\Operation\Request\RequestInterface; +use FreeDSx\Ldap\Operation\Request\SearchRequest; use FreeDSx\Ldap\Operation\Response\DeleteResponse; +use FreeDSx\Ldap\Operation\Response\SearchResultDone; +use FreeDSx\Ldap\Operation\Response\SearchResultEntry; use FreeDSx\Ldap\Operation\ResultCode; use FreeDSx\Ldap\Protocol\LdapMessageRequest; use FreeDSx\Ldap\Protocol\LdapMessageResponse; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseStream; +use FreeDSx\Ldap\Protocol\Queue\Response\ResponseWriter; use FreeDSx\Ldap\Protocol\Queue\ServerQueue; +use FreeDSx\Ldap\Search\Filters; use FreeDSx\Ldap\Server\AccessControl\AccessControlInterface; use FreeDSx\Ldap\Server\Backend\LdapBackendInterface; -use FreeDSx\Ldap\Server\Middleware\OperationErrorMiddleware; use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext; +use FreeDSx\Ldap\Server\Middleware\ResponseWriterMiddleware; use FreeDSx\Ldap\Server\Operation\OperationOutcome; use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult; +use FreeDSx\Ldap\Server\Operation\SearchOperationResult; use FreeDSx\Ldap\Server\Token\TokenInterface; +use Generator; use PHPUnit\Framework\MockObject\MockObject; use PHPUnit\Framework\TestCase; +use Tests\Support\FreeDSx\Ldap\Middleware\StreamMiddlewareHandler; use Tests\Support\FreeDSx\Ldap\Middleware\StubMiddlewareHandler; use Tests\Support\FreeDSx\Ldap\Middleware\ThrowingMiddlewareHandler; -final class OperationErrorMiddlewareTest extends TestCase +final class ResponseWriterMiddlewareTest extends TestCase { private ServerQueue&MockObject $mockQueue; @@ -45,9 +54,12 @@ final class OperationErrorMiddlewareTest extends TestCase private TokenInterface&MockObject $mockToken; - private OperationErrorMiddleware $subject; + private ResponseWriterMiddleware $subject; - private ?LdapMessageResponse $sent = null; + /** + * @var list + */ + private array $sent = []; protected function setUp(): void { @@ -57,40 +69,46 @@ protected function setUp(): void $this->mockToken = $this->createMock(TokenInterface::class); $this->mockQueue - ->method('sendMessage') - ->willReturnCallback(function (LdapMessageResponse $response): ServerQueue { - $this->sent = $response; + ->method('sendMessages') + ->willReturnCallback(function (iterable $responses): ServerQueue { + foreach ($responses as $response) { + if ($response instanceof LdapMessageResponse) { + $this->sent[] = $response; + } + } return $this->mockQueue; }); - $this->subject = new OperationErrorMiddleware( - $this->mockQueue, + $this->subject = new ResponseWriterMiddleware( + new ResponseWriter($this->mockQueue), $this->mockBackend, $this->mockAccessControl, ); } - public function test_it_passes_a_successful_result_through_without_sending(): void + public function test_it_resolves_a_successful_outcome_without_rendering_an_error(): void { - $this->mockQueue - ->expects(self::never()) - ->method('sendMessage'); - $result = OperationOutcomeResult::succeeded(); + $stream = $this->subject->process( + $this->contextFor(new DeleteRequest('cn=foo,dc=bar')), + new StubMiddlewareHandler($result), + ); + self::assertSame( $result, - $this->subject->process( - $this->contextFor(new DeleteRequest('cn=foo,dc=bar')), - new StubMiddlewareHandler($result), - ), + $stream->outcome(), + ); + self::assertSame( + [], + $this->sent, ); } public function test_it_renders_the_response_for_a_thrown_exception_and_does_not_rethrow(): void { - $result = $this->subject->process( + $stream = $this->subject->process( $this->contextFor(new DeleteRequest('cn=foo,dc=bar')), new ThrowingMiddlewareHandler(new OperationException( 'Unwilling.', @@ -100,10 +118,10 @@ public function test_it_renders_the_response_for_a_thrown_exception_and_does_not self::assertSame( OperationOutcome::Failed, - $result->outcome(), + $stream->outcome()->outcome(), ); - $response = $this->sent?->getResponse(); + $response = $this->firstSent()?->getResponse(); self::assertInstanceOf(DeleteResponse::class, $response); self::assertSame( ResultCode::UNWILLING_TO_PERFORM, @@ -111,6 +129,35 @@ public function test_it_renders_the_response_for_a_thrown_exception_and_does_not ); } + public function test_a_mid_stream_failure_is_answered_after_the_entries_already_written(): void + { + $entry = new LdapMessageResponse(1, new SearchResultEntry(Entry::create('cn=a,dc=bar'))); + + $this->subject->process( + $this->contextFor((new SearchRequest(Filters::present('cn')))->base('dc=bar')), + new StreamMiddlewareHandler(ResponseStream::streaming( + (function () use ($entry): Generator { + yield $entry; + + throw new OperationException('Backend failed.', ResultCode::OPERATIONS_ERROR); + })(), + static fn(): SearchOperationResult => throw new OperationException('unreached'), + )), + ); + + // The already-produced entry went out, then the failure rendered as the search's terminal. + self::assertSame( + $entry, + $this->sent[0] ?? null, + ); + $done = $this->sent[1]->getResponse() ?? null; + self::assertInstanceOf(SearchResultDone::class, $done); + self::assertSame( + ResultCode::OPERATIONS_ERROR, + $done->getResultCode(), + ); + } + public function test_it_keeps_a_matched_dn_the_token_may_see(): void { $this->mockBackend @@ -125,7 +172,7 @@ public function test_it_keeps_a_matched_dn_the_token_may_see(): void new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))), ); - $response = $this->sent?->getResponse(); + $response = $this->firstSent()?->getResponse(); self::assertInstanceOf(DeleteResponse::class, $response); self::assertSame( 'dc=bar', @@ -147,7 +194,7 @@ public function test_it_drops_a_matched_dn_the_access_control_hides(): void new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))), ); - $response = $this->sent?->getResponse(); + $response = $this->firstSent()?->getResponse(); self::assertInstanceOf(DeleteResponse::class, $response); self::assertSame( '', @@ -166,7 +213,7 @@ public function test_it_drops_a_matched_dn_when_the_backend_has_no_ancestor(): v new ThrowingMiddlewareHandler($this->noSuchObject(new Dn('dc=bar'))), ); - $response = $this->sent?->getResponse(); + $response = $this->firstSent()?->getResponse(); self::assertInstanceOf(DeleteResponse::class, $response); self::assertSame( '', @@ -184,6 +231,11 @@ private function noSuchObject(Dn $matchedDn): OperationException ); } + private function firstSent(): ?LdapMessageResponse + { + return $this->sent[0] ?? null; + } + private function contextFor(RequestInterface $request): ServerRequestContext { return new ServerRequestContext( diff --git a/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php b/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php index 5b15ca63..846be038 100644 --- a/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php +++ b/tests/unit/Server/Proxy/ProxyRequestForwarderTest.php @@ -65,7 +65,7 @@ public function test_it_relays_a_single_response_under_the_original_message_id() new DeleteRequest('cn=foo,dc=bar'), )); - self::assertSame(OperationOutcome::Succeeded, $result->outcome()); + self::assertSame(OperationOutcome::Succeeded, $result->outcome()->outcome()); } public function test_it_relays_an_upstream_error_as_a_response(): void @@ -92,7 +92,7 @@ public function test_it_relays_an_upstream_error_as_a_response(): void new DeleteRequest('cn=missing,dc=bar'), )); - self::assertSame(OperationOutcome::Failed, $result->outcome()); + self::assertSame(OperationOutcome::Failed, $result->outcome()->outcome()); } public function test_it_closes_both_connections_on_unbind(): void @@ -121,7 +121,7 @@ public function test_it_does_not_forward_an_abandon(): void new AbandonRequest(2), )); - self::assertSame(OperationOutcome::Succeeded, $result->outcome()); + self::assertSame(OperationOutcome::Succeeded, $result->outcome()->outcome()); } private function contextFor( diff --git a/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php b/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php index fecbd0e7..4cd33274 100644 --- a/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php +++ b/tests/unit/Server/Proxy/ProxyRequestPipelineTest.php @@ -59,7 +59,7 @@ public function test_it_forwards_everything_else(): void $this->forwarder ->expects(self::once()) ->method('handle') - ->willReturn(OperationOutcomeResult::succeeded()); + ->willReturn(ResponseStream::none(OperationOutcomeResult::succeeded())); $this->startTlsHandler ->expects(self::never()) ->method('handleRequest');