[1.0.0] Consolidate server queue writing to one spot (Part 2)

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(),

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,

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);
+    }
 }

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<LdapMessageResponse> ...$streams
+     * @return Generator<LdapMessageResponse>
+     */
+    private function concat(Generator ...$streams): Generator
+    {
+        foreach ($streams as $stream) {
+            yield from $stream;
+        }
     }
 
     /**

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;
@@ -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,
@@ -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),
         );
     }
 }

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();
 

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()) {

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());
     }
 }

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(),

src/FreeDSx/Ldap/Server/Middleware/MetricsMiddleware.php

@@ -21,14 +21,14 @@
 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;
 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\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(