Extend opaque correlation to single-key writes

CHANGELOG.md

@@ -4,6 +4,11 @@ Dalli Changelog
 Unreleased
 ==========
 
+- Extend opt-in opaque correlation to single-key writes: `set`, `add`, `replace`, `append`, `prepend`, `incr`,
+  `decr`, and the writes made by `fetch`/`cas`/`cas!`. Rejected replies return `false` (`nil` for `incr`/`decr`) and
+  recycle the connection, with the same log and trace diagnostics as reads. Deletes, quiet blocks, and multi-key
+  writes are unchanged. (mrattle)
+
 - Read all non-pipelined responses through `Dalli::Protocol::BufferedReader`: reads wait with explicit deadlines and
   poll the socket once more before timing out, so replies delayed only by a busy fiber scheduler are not discarded.
   The reader holds one buffer per connection that grows from 8 KiB to 64 KiB once a reply outgrows it, reads up to

README.md

@@ -27,28 +27,31 @@ The name is a variant of Salvador Dali for his famous painting [The Persistence
 
 ## Optional response correlation
 
-Enable single-get response validation with `Dalli::Client.new(servers, correlate_with_opaques: true)`.
-Dalli uses the first caller-supplied `O` token in `meta_flags`, or generates a four-character (24-bit) token if none is supplied.
-It sends that token exactly once and checks the echoed value against it.
-
-Mismatched or missing tokens on accepted get replies return a miss and close the connection, without retrying or down-marking.
-Other protocol errors are unchanged. Omit the option or set it to `false` to leave caller opaques and default reads unchanged.
+Enable single-key response validation with `Dalli::Client.new(servers, correlate_with_opaques: true)`.
+Reads use the first caller-supplied `O` token in `meta_flags`, or generate a four-character (24-bit) token if none is supplied;
+writes always generate one. Dalli sends that token exactly once and checks the echoed value against it.
+
+Mismatched or missing tokens on accepted replies close the connection, without retrying or down-marking.
+Reads return a miss. `set`, `add`, `replace`, `append`, and `prepend` return `false`, and `incr`/`decr` return `nil`;
+the server may still have applied a rejected write.
+Other protocol errors are unchanged. Omit the option or set it to `false` to leave caller opaques and default requests unchanged.
 Caller tokens must be protocol-safe and suitably unique per request; reused tokens cannot detect swaps between those requests.
-Multi-get/pipeline behavior is unchanged.
+Deletes, `quiet`/`multi` blocks, and multi-key operations such as `get_multi`, `set_multi`, and `delete_multi` are unchanged.
 
 For opt-in, per-request opaque correlation, also set `opaque_correlation_request_only: true` on the client and pass `correlate_with_opaques: true` in request options.  The PRNG is still initialized at connection setup, but unflagged requests retain their baseline behavior:
 
 ```ruby
 client = Dalli::Client.new(servers, correlate_with_opaques: true, opaque_correlation_request_only: true)
 client.get('key')                             # no correlation
 client.get('key', correlate_with_opaques: true) # correlate this read
+client.set('key', 'value', nil, correlate_with_opaques: true) # correlate this write
 ```
 
 Request-only mode defaults to false. Requests can explicitly opt out with `correlate_with_opaques: false`
 in either mode, but cannot enable correlation when the client option is false or omitted.
 An omitted or `nil` request option inherits the client mode.
-Request options do not mutate client configuration. They also apply to `gat`, CAS reads,
-`get_with_status`, `touch`, and the reads made by `fetch`/`cas!`.
+Request options do not mutate client configuration. They also apply to `gat`, CAS reads, `get_with_status`, `touch`,
+`add`, `replace`, `append`, `prepend`, `incr`, `decr`, and the reads and writes made by `fetch`/`cas`/`cas!`.
 
 ## Timeout recovery under fiber schedulers
 

lib/dalli/client.rb

@@ -29,8 +29,10 @@ class Client
     # - :namespace - prepend each key with this value to provide simple namespacing.
     # - :failover - if a server is down, look for and store values on another server in the ring.  Default: true.
     # - :threadsafe - ensure that only one thread is actively using a socket at a time. Default: true.
-    # - :correlate_with_opaques - validate single-get replies against the first caller O token, or a generated token.
-    #                           Defaults to false; an enabled client can opt individual reads out of correlation.
+    # - :correlate_with_opaques - validate single-key get and write replies against a request O token: a read's first
+    #                           caller O token in meta_flags, or a generated token. Deletes, quiet blocks, and
+    #                           multi-key operations are not correlated. Defaults to false; an enabled client can opt
+    #                           individual requests out of correlation.
     # - :opaque_correlation_request_only - require per-request opt-in on a correlation-enabled client.
     #                                     Defaults to false; does not change eager PRNG initialization.
     # - :expires_in - default TTL in seconds if you do not pass TTL as a parameter to an individual operation, defaults

lib/dalli/middlewares.rb

@@ -13,7 +13,7 @@ def retrieve_req(_operation, _tags = {})
       yield attributes
     end
 
-    # Diagnostic hooks run synchronously in retrieve_req, under the server lock when threadsafe.
+    # Diagnostic hooks run synchronously in retrieve_req and storage_req, under the server lock when threadsafe.
     # With threadsafe: false, callers must provide exclusive client access.
     def record_request_opaque(_opaque); end
 

lib/dalli/opentelemetry_middleware.rb

@@ -11,41 +11,21 @@ module OpentelemetryMiddleware
       'db.system' => 'memcached'
     }.freeze
 
-    def storage_req(operation, tags = {})
-      TRACER.in_span(operation, attributes: tags.merge!(DEFAULT_TRACE_ATTRIBUTES), kind: :client) do |span|
-        attributes = {}
-        result = yield attributes
-        span.add_attributes(attributes)
-        result
-      end
+    def storage_req(operation, tags = {}, &)
+      traced_request(operation, tags, &)
     end
 
-    # Record computed attributes only on normal completion, matching the other hooks.
-    def retrieve_req(operation, tags = {})
-      TRACER.in_span(operation, attributes: tags.merge!(DEFAULT_TRACE_ATTRIBUTES), kind: :client) do |span|
-        attributes = {}
-        trace_attributes = {}
-        # Preserve outer state for reentrant instrumentation, not overlapping socket requests.
-        previous_trace_attributes = @retrieval_trace_attributes
-        begin
-          @retrieval_trace_attributes = trace_attributes
-          result = yield attributes
-          span.add_attributes(attributes)
-          span.add_attributes(trace_attributes) unless trace_attributes.empty?
-          result
-        ensure
-          @retrieval_trace_attributes = previous_trace_attributes
-        end
-      end
+    def retrieve_req(operation, tags = {}, &)
+      traced_request(operation, tags, &)
     end
 
     # Keep unique tokens out of metadata shared with general metrics middleware.
     def record_request_opaque(opaque)
-      @retrieval_trace_attributes['request_opaque'] = opaque if @retrieval_trace_attributes
+      @request_trace_attributes['request_opaque'] = opaque if @request_trace_attributes
     end
 
     def correlation_failure(attributes)
-      @retrieval_trace_attributes&.merge!(attributes)
+      @request_trace_attributes&.merge!(attributes)
     end
 
     def storage_req_pipeline(operation, tags = {})
@@ -65,5 +45,26 @@ def retrieve_req_pipeline(operation, tags = {})
         result
       end
     end
+
+    private
+
+    # Record computed attributes only on normal completion, matching the other hooks.
+    def traced_request(operation, tags)
+      TRACER.in_span(operation, attributes: tags.merge!(DEFAULT_TRACE_ATTRIBUTES), kind: :client) do |span|
+        attributes = {}
+        trace_attributes = {}
+        # Preserve outer state for reentrant instrumentation, not overlapping socket requests.
+        previous_trace_attributes = @request_trace_attributes
+        begin
+          @request_trace_attributes = trace_attributes
+          result = yield attributes
+          span.add_attributes(attributes)
+          span.add_attributes(trace_attributes) unless trace_attributes.empty?
+          result
+        ensure
+          @request_trace_attributes = previous_trace_attributes
+        end
+      end
+    end
   end
 end

lib/dalli/protocol/meta.rb

@@ -42,6 +42,11 @@ def select_request_opaque(request_options:, meta_flags: nil)
         @connection_manager.generate_opaque
       end
 
+      # Quiet writes suppress success replies, leaving nothing to correlate.
+      def select_write_opaque(options)
+        select_request_opaque(request_options: options) unless quiet?
+      end
+
       def record_request_opaque(opaque)
         @middlewares_stack.record_request_opaque(opaque) if opaque
       end
@@ -407,50 +412,48 @@ def do_storage_req(mode, key, raw_value, ttl = nil, cas = nil, options = {})
         (value, bitflags) = @value_marshaller.store(key, raw_value, options)
         ttl = TtlSanitizer.sanitize(ttl) if ttl
         encoded_key, base64 = KeyRegularizer.encode(key)
+        opaque = select_write_opaque(options)
         @middlewares_stack.storage_req(
           "memcached.#{mode}",
           { 'keys' => key, 'value_size' => value.bytesize, 'ttl' => ttl }
         ) do
-          req = RequestFormatter.meta_set(
-            key: encoded_key,
-            value: value,
-            bitflags: bitflags,
-            cas: cas,
-            ttl: ttl,
-            mode: mode,
-            quiet: quiet?,
-            base64: base64,
-            **routing_token_kwargs(options)
-          )
+          record_request_opaque(opaque)
+          req = RequestFormatter.meta_set(key: encoded_key, value: value, bitflags: bitflags, cas: cas, ttl: ttl,
+                                          mode: mode, quiet: quiet?, base64: base64, opaque: opaque,
+                                          **routing_token_kwargs(options))
           write(req)
           write(value)
           write(TERMINATOR)
           @connection_manager.flush
 
-          response_processor.meta_set_with_cas unless quiet?
+          response_processor.meta_set_with_cas(expected_opaque: opaque) unless quiet?
         end
       end
 
       def append(key, value, options = nil)
+        opaque = select_write_opaque(options)
         @middlewares_stack.storage_req('memcached.append', { 'keys' => key, 'value_size' => value.bytesize }) do
-          write_append_prepend_req(:append, key, value, nil, nil, options || {})
-          response_processor.meta_set_append_prepend unless quiet?
+          record_request_opaque(opaque)
+          write_append_prepend_req(:append, key, value, nil, nil, options || {}, opaque: opaque)
+          response_processor.meta_set_append_prepend(expected_opaque: opaque) unless quiet?
         end
       end
 
       def prepend(key, value, options = nil)
+        opaque = select_write_opaque(options)
         @middlewares_stack.storage_req('memcached.prepend', { 'keys' => key, 'value_size' => value.bytesize }) do
-          write_append_prepend_req(:prepend, key, value, nil, nil, options || {})
-          response_processor.meta_set_append_prepend unless quiet?
+          record_request_opaque(opaque)
+          write_append_prepend_req(:prepend, key, value, nil, nil, options || {}, opaque: opaque)
+          response_processor.meta_set_append_prepend(expected_opaque: opaque) unless quiet?
         end
       end
 
       # rubocop:disable-next Metrics/ParameterLists
-      def write_append_prepend_req(mode, key, value, ttl = nil, cas = nil, options = {})
+      def write_append_prepend_req(mode, key, value, ttl = nil, cas = nil, options = {}, opaque: nil)
         ttl = TtlSanitizer.sanitize(ttl) if ttl
         encoded_key, base64 = KeyRegularizer.encode(key)
         req = RequestFormatter.meta_set(key: encoded_key, value: value, base64: base64,
-                                        cas: cas, ttl: ttl, mode: mode, quiet: quiet?,
+                                        cas: cas, ttl: ttl, mode: mode, quiet: quiet?, opaque: opaque,
                                         **routing_token_kwargs(options))
         write(req)
         write(value)
@@ -485,6 +488,7 @@ def incr(key, count, ttl, initial, options = nil)
       def decr_incr(incr, key, delta, ttl, initial, options = nil)
         ttl = initial ? TtlSanitizer.sanitize(ttl) : nil # Only set a TTL if we want to set a value on miss
         encoded_key, base64 = KeyRegularizer.encode(key)
+        opaque = select_write_opaque(options)
 
         @middlewares_stack.storage_req(
           incr ? 'memcached.incr' : 'memcached.decr',
@@ -495,11 +499,12 @@ def decr_incr(incr, key, delta, ttl, initial, options = nil)
             'initial' => initial || 0
           }
         ) do
+          record_request_opaque(opaque)
           write(RequestFormatter.meta_arithmetic(key: encoded_key, delta: delta, initial: initial, incr: incr, ttl: ttl,
-                                                 quiet: quiet?, base64: base64,
+                                                 quiet: quiet?, base64: base64, opaque: opaque,
                                                  **routing_token_kwargs(options)))
           @connection_manager.flush
-          response_processor.decr_incr unless quiet?
+          response_processor.decr_incr(expected_opaque: opaque) unless quiet?
         end
       end
 

lib/dalli/protocol/meta/request_formatter.rb

@@ -36,7 +36,7 @@ def self.meta_get(key:, opaque: nil, value: true, return_cas: false, ttl: nil, b
         end
 
         def self.meta_set(key:, value:, bitflags: nil, cas: nil, ttl: nil, mode: :set, base64: false, quiet: false,
-                          p_token: nil, l_token: nil)
+                          opaque: nil, p_token: nil, l_token: nil)
           cmd = "ms #{key} #{value.bytesize}"
           cmd << ' c' if !quiet && !%i[append prepend].include?(mode)
           cmd << ' b' if base64
@@ -45,6 +45,7 @@ def self.meta_set(key:, value:, bitflags: nil, cas: nil, ttl: nil, mode: :set, b
           cmd << " T#{ttl}" if ttl
           cmd << " M#{mode_to_token(mode)}"
           cmd << ' q' if quiet
+          cmd << " O#{opaque}" if opaque
           cmd << routing_tokens(p_token: p_token, l_token: l_token)
           cmd << TERMINATOR
         end
@@ -66,7 +67,7 @@ def self.meta_delete(key:, cas: nil, ttl: nil, base64: false, quiet: false, p_to
         end
 
         def self.meta_arithmetic(key:, delta:, initial:, incr: true, cas: nil, ttl: nil, base64: false, quiet: false,
-                                 p_token: nil, l_token: nil)
+                                 opaque: nil, p_token: nil, l_token: nil)
           cmd = "ma #{key} v"
           cmd << ' b' if base64
           cmd << " D#{delta}" if delta
@@ -76,6 +77,7 @@ def self.meta_arithmetic(key:, delta:, initial:, incr: true, cas: nil, ttl: nil,
           cmd << cas_string(cas)
           cmd << ' q' if quiet
           cmd << " M#{incr ? 'I' : 'D'}"
+          cmd << " O#{opaque}" if opaque
           cmd << routing_tokens(p_token: p_token, l_token: l_token)
           cmd + TERMINATOR
         end

lib/dalli/protocol/meta/response_processor.rb

@@ -93,15 +93,17 @@ def meta_get_with_status(expected_opaque: nil)
           end
         end
 
-        def meta_set_with_cas
+        def meta_set_with_cas(expected_opaque: nil)
           tokens = error_on_unexpected!([HD, NS, NF, EX])
+          return false unless verify_opaque!(tokens, expected_opaque)
           return false unless tokens.first == HD
 
           cas_from_tokens(tokens)
         end
 
-        def meta_set_append_prepend
+        def meta_set_append_prepend(expected_opaque: nil)
           tokens = error_on_unexpected!([HD, NS, NF, EX])
+          return false unless verify_opaque!(tokens, expected_opaque)
           return false unless tokens.first == HD
 
           true
@@ -112,8 +114,9 @@ def meta_delete
           tokens.first == HD
         end
 
-        def decr_incr
+        def decr_incr(expected_opaque: nil)
           tokens = error_on_unexpected!([VA, NF, NS, EX])
+          return nil unless verify_opaque!(tokens, expected_opaque)
           return false if [NS, EX].include?(tokens.first)
           return nil if tokens.first == NF