closes trailers eagerly to allow pure header responses to be publishe…

…d asap

waiter/integration/waiter/grpc_test.clj

@@ -228,7 +228,7 @@
                       ids (map #(str "id-" (cond-> % (= % error-index) (str "." mode))) (range num-messages))
                       messages (doall (repeatedly num-messages #(rand-str (inc (rand-int max-message-length)))))
                       request-headers (assoc request-headers "x-cid" collect-cid)
-                      _ (println "collect packages cid" collect-cid "for"
+                      _ (log/info "collect packages cid" collect-cid "for"
                                   {:iteration iteration :max-message-length max-message-length})
                       summaries (GrpcClient/collectPackages
                                   host h2c-port request-headers ids from messages 1 true (inc num-messages))
@@ -257,17 +257,6 @@
                   (is status-summary assertion-message)
                   (when status-summary
                     (log/info "server cancellation summary" status-summary)
-                    (if (zero? iteration)
-                      (do
-                        ;; TODO this if-block should not be needed, cancellation should be propagated correctly to the client
-                        (is (= "INTERNAL" (.getStatusCode status-summary))
-                            assertion-message)
-                        (is (str/includes? (.getStatusDescription status-summary) "Received Rst Stream")
-                            assertion-message))
-                      (do
-                        (is (= "CANCELLED" (.getStatusCode status-summary))
-                            assertion-message)
-                        (is (= "Cancelled by server" (.getStatusDescription status-summary))
-                            assertion-message)))
-                    (is (zero? (.getNumMessages status-summary))
-                        assertion-message)))))))))))
+                    (is (= "CANCELLED" (.getStatusCode status-summary)) assertion-message)
+                    (is (= "Cancelled by server" (.getStatusDescription status-summary)) assertion-message)
+                    (is (zero? (.getNumMessages status-summary)) assertion-message)))))))))))

waiter/src/waiter/process_request.clj

@@ -524,7 +524,6 @@
               (try
                 (let [trailers-map (async/<! trailers)
                       modified-trailers (merge grpc-headers trailers-map)]
-                  (log/info "response trailers:" trailers-map)
                   (when (seq grpc-headers)
                     (log/info "attaching grpc headers into trailer:" grpc-headers))
                   (when (seq modified-trailers)
@@ -541,7 +540,7 @@
   "Eagerly terminates grpc requests with error status headers.
    We cannot rely on jetty to close the request for us in a timely manner,
    please see https://github.com/eclipse/jetty.project/issues/3842 for details."
-  [{:keys [abort-ch body error-chan] :as response} request backend-proto reservation-status-promise]
+  [{:keys [abort-ch body error-chan trailers] :as response} request backend-proto reservation-status-promise]
   (let [request-headers (:headers request)
         {:strs [grpc-status]} (:headers response)
         proto-version (hu/backend-protocol->http-version backend-proto)]
@@ -550,13 +549,14 @@
                (not= "0" grpc-status)
                (au/chan? body))
       (log/info "eagerly closing response body as grpc status is" grpc-status)
+      ;; mark the request as successful, grpc failures are reported in the headers
+      (deliver reservation-status-promise :success)
       (when abort-ch
         ;; disallow aborting the request
         (async/close! abort-ch))
-      ;; mark the request as successful, grpc failures are reported in the headers
-      (deliver reservation-status-promise :success)
       ;; stop writing any content in the body
       (async/close! body)
+      (async/close! trailers)
       (async/close! error-chan))
     response))
 
@@ -593,8 +593,8 @@
             location (post-process-async-request-response-fn
                        service-id metric-group backend-proto instance (handler/make-auth-user-map request)
                        reason-map instance-request-properties location query-string))
-          (forward-grpc-status-headers-in-trailers)
           (handle-grpc-response request backend-proto reservation-status-promise)
+          (forward-grpc-status-headers-in-trailers)
           (assoc :body resp-chan)
           (update-in [:headers] (fn update-response-headers [headers]
                                   (utils/filterm #(not= "connection" (str/lower-case (str (key %)))) headers)))))))