Raise pipelined meta protocol errors after draining

lib/dalli/protocol/base.rb

@@ -96,6 +96,7 @@ def unlock!; end
       # Returns nothing.
       def pipeline_response_setup
         verify_pipelined_state(:getkq)
+        @pipeline_error = nil
         write_noop
         response_buffer.reset
       end
@@ -112,20 +113,20 @@ def pipeline_next_responses
 
         response_buffer.read
 
-        status, cas, key, value = response_buffer.process_single_getk_response
         # status is not nil only if we have a full response to parse
         # in the buffer
-        until status.nil?
-          # If the status is ok and key is nil, then this is the response
-          # to the noop at the end of the pipeline
-          finish_pipeline && break if status && key.nil?
+        loop do
+          response = response_buffer.process_single_getk_response
+          break if response.first.nil?
 
-          # If the status is ok and the key is not nil, then this is a
-          # getkq response with a value that we want to set in the response hash
-          values[key] = [value, cas] unless key.nil?
+          process_pipeline_getk_response(values, response)
+          break if pipeline_complete?
+        end
 
-          # Get the next response from the buffer
-          status, cas, key, value = response_buffer.process_single_getk_response
+        if pipeline_complete? && @pipeline_error
+          error = @pipeline_error
+          @pipeline_error = nil
+          raise error
         end
 
         values
@@ -138,7 +139,23 @@ def pipeline_next_responses
       # disconnected, and the exception is swallowed.
       #
       # Returns nothing.
+      def process_pipeline_getk_response(values, response)
+        status, cas, key, value, error = response
+        @pipeline_error ||= error if error
+
+        # If the status is ok and key is nil, then this is the response
+        # to the noop at the end of the pipeline
+        if status && key.nil?
+          finish_pipeline
+        elsif !key.nil?
+          # If the status is ok and the key is not nil, then this is a
+          # getkq response with a value that we want to set in the response hash
+          values[key] = [value, cas]
+        end
+      end
+
       def pipeline_abort
+        @pipeline_error = nil
         response_buffer.clear
         @connection_manager.abort_request!
         return true unless connected?

lib/dalli/protocol/meta.rb

@@ -82,12 +82,18 @@ def read_multi_req(keys, req_options = nil)
           @connection_manager.flush
 
           terminator_length = TERMINATOR.length
+          error = nil
           while (line = @connection_manager.readline)
             break if line == TERMINATOR || line[0, 2] == 'MN'
+
+            tokens = line.split
+            if response_processor.error_response?(tokens.first)
+              error ||= response_processor.response_error_from_line(line.chomp(TERMINATOR))
+              next
+            end
             next unless line[0, 3] == 'VA '
 
             # VA value_length flags key
-            tokens = line.split
             value = @connection_manager.read_exact(tokens[1].to_i)
             bitflags = optimized_for_raw ? 0 : @response_processor.bitflags_from_tokens(tokens)
             @connection_manager.read_exact(terminator_length) # read the terminator
@@ -103,6 +109,8 @@ def read_multi_req(keys, req_options = nil)
             attributes['hit_count'] = results.size
             attributes['miss_count'] = keys.size - results.size
           end
+
+          raise error if error
         end
 
         results
@@ -131,11 +139,17 @@ def read_multi_with_status_req(keys, req_options = nil)
           @connection_manager.flush
 
           terminator_length = TERMINATOR.length
+          error = nil
           while (line = @connection_manager.readline)
             break if line == TERMINATOR || line[0, 2] == 'MN'
-            next unless line[0, 3] == 'VA '
 
             tokens = line.split
+            if response_processor.error_response?(tokens.first)
+              error ||= response_processor.response_error_from_line(line.chomp(TERMINATOR))
+              next
+            end
+            next unless line[0, 3] == 'VA '
+
             value = @connection_manager.read_exact(tokens[1].to_i)
             bitflags = optimized_for_raw ? 0 : response_processor.bitflags_from_tokens(tokens)
             @connection_manager.read_exact(terminator_length)
@@ -167,6 +181,8 @@ def read_multi_with_status_req(keys, req_options = nil)
             attributes['miss_count'] = keys.size - fresh_hit_count
             attributes['stale_count'] = stale_count
           end
+
+          raise error if error
         end
 
         results

lib/dalli/protocol/meta/response_processor.rb

@@ -21,7 +21,10 @@ class ResponseProcessor
         STAT = 'STAT'
         VA = 'VA'
         VERSION = 'VERSION'
+        CLIENT_ERROR = 'CLIENT_ERROR'
+        ERROR = 'ERROR'
         SERVER_ERROR = 'SERVER_ERROR'
+        ERROR_RESPONSES = [CLIENT_ERROR, SERVER_ERROR, ERROR].freeze
 
         def initialize(io_source, value_marshaller)
           @io_source = io_source
@@ -141,25 +144,43 @@ def version
         end
 
         def consume_all_responses_until_mn
-          tokens = next_line_to_tokens
+          error = nil
+
+          each_response_until_mn do |line, tokens|
+            error ||= response_error_from_line(line) if error_response?(tokens.first)
+          end
+
+          raise error if error
 
-          tokens = next_line_to_tokens while tokens.first != MN
           true
         end
 
         # In quiet mode, only error responses (NF) are sent, success (HD) is suppressed.
         # Returns the count of NF (not found) responses.
         def count_not_found_responses_until_mn
+          error = nil
           not_found_count = 0
-          tokens = next_line_to_tokens
 
-          while tokens.first != MN
+          each_response_until_mn do |line, tokens|
+            error ||= response_error_from_line(line) if error_response?(tokens.first)
             not_found_count += 1 if tokens.first == NF
-            tokens = next_line_to_tokens
           end
+
+          raise error if error
+
           not_found_count
         end
 
+        def each_response_until_mn
+          loop do
+            line = read_line
+            tokens = line&.split || []
+            break if tokens.first == MN
+
+            yield line, tokens
+          end
+        end
+
         def tokens_from_header_buffer(buf)
           header = header_from_buffer(buf)
           tokens = header.split
@@ -170,7 +191,7 @@ def tokens_from_header_buffer(buf)
 
         def full_response_from_buffer(tokens, body, resp_size)
           value = @value_marshaller.retrieve(body, bitflags_from_tokens(tokens))
-          [resp_size, tokens.first == VA, cas_from_tokens(tokens), key_from_tokens(tokens), value]
+          [resp_size, tokens.first == VA, cas_from_tokens(tokens), key_from_tokens(tokens), value, nil]
         end
 
         ##
@@ -185,20 +206,26 @@ def full_response_from_buffer(tokens, body, resp_size)
         ##
         def getk_response_from_buffer(buf)
           # There's no header in the buffer, so don't advance
-          return [0, nil, nil, nil, nil] unless contains_header?(buf)
+          return [0, nil, nil, nil, nil, nil] unless contains_header?(buf)
 
           tokens, header_len, body_len = tokens_from_header_buffer(buf)
 
-          # We have a complete response that has no body.
-          # This is either the response to the terminating
-          # noop or, if the status is not MN, an intermediate
-          # error response that needs to be discarded.
-          return [header_len, true, nil, nil, nil] if body_len.zero?
+          # Error responses from an upstream proxy/partition are explicitly
+          # discarded and must not be returned to callers as raw values or
+          # associated with any requested key.
+          if error_response?(tokens.first)
+            return [header_len, false, nil, nil, nil, response_error_from_line(header_from_buffer(buf))]
+          end
+
+          # We have a complete response that has no body.  Only MN is the
+          # response to the terminating noop; other body-less responses are
+          # discarded.
+          return [header_len, tokens.first == MN, nil, nil, nil, nil] if body_len.zero?
 
           resp_size = header_len + body_len + TERMINATOR.length
           # The header is in the buffer, but the body is not.  As we don't have
           # a complete response, don't advance the buffer
-          return [0, nil, nil, nil, nil] unless buf.bytesize >= resp_size
+          return [0, nil, nil, nil, nil, nil] unless buf.bytesize >= resp_size
 
           # The full response is in our buffer, so parse it and return
           # the values
@@ -220,11 +247,22 @@ def error_on_unexpected!(expected_codes)
 
           return tokens if expected_codes.include?(tokens.first)
 
-          raise Dalli::ServerError, line if tokens.first == SERVER_ERROR
+          raise response_error_from_line(line) if error_response?(tokens.first)
 
           raise Dalli::DalliError, "Response error: #{line}"
         end
 
+        def response_error_from_line(line)
+          tokens = line&.split || []
+          return Dalli::ServerError.new(line) if tokens.first == SERVER_ERROR
+
+          Dalli::DalliError.new("Response error: #{line}")
+        end
+
+        def error_response?(code)
+          ERROR_RESPONSES.include?(code)
+        end
+
         def meta_flags_from_tokens(tokens)
           {
             c: cas_from_tokens(tokens),

lib/dalli/protocol/response_buffer.rb

@@ -22,9 +22,9 @@ def read
       # Attempts to process a single response from the buffer.  Starts
       # by advancing the buffer to the specified start position
       def process_single_getk_response
-        bytes, status, cas, key, value = @response_processor.getk_response_from_buffer(@buffer)
+        bytes, status, cas, key, value, error = @response_processor.getk_response_from_buffer(@buffer)
         advance(bytes)
-        [status, cas, key, value]
+        [status, cas, key, value, error]
       end
 
       # Advances the internal response buffer by bytes_to_advance