fix: Reply within async_dispatch block in squashed pipeline

Signed-off-by: Vladislav Oleshko <[email protected]>

src/facade/dragonfly_connection.cc

@@ -854,16 +854,19 @@ void Connection::DispatchFiber(util::FiberSocketBase* peer) {
       service_->DispatchManyCommands(absl::MakeSpan(args), cc_.get());
       cc_->async_dispatch = false;
 
+      // Flush strictly before the dispatch queue is cleared so that no sync dispatch can occur
+      if (dispatch_q_.size() == args.size())  // Flush if no new messages appeared
+        builder->FlushBatch();
+
+      builder->SetBatchMode(false);  // in case the next dispatch is sync
+      DCHECK(!cc_->sync_dispatch);
+
       // Dispatch queue could have grown, so handle strictly as many as we executed
       for (size_t i = 0; i < args.size(); i++) {
         recycle(move(dispatch_q_.front()));
         dispatch_q_.pop_front();
       }
 
-      if (dispatch_q_.empty()) {
-        builder->FlushBatch();
-        builder->SetBatchMode(false);  // in case the next dispatch is sync
-      }
     } else {
       MessageHandle msg = move(dispatch_q_.front());
       dispatch_q_.pop_front();