From fa53cde2359bc6f618bf44dc855c7f357e29270d Mon Sep 17 00:00:00 2001
From: Mark Rattle <mark.rattle@shopify.com>
Date: Fri, 23 Oct 2026 16:33:36 -0400
Subject: [PATCH] Extend opaque correlation to single-key writes

Send a generated O token on non-quiet ms/ma requests (set, add, replace, append, prepend, incr, decr, and the writes made by fetch/cas/cas!) and verify the echoed token before reading anything else from the reply. Rejected replies return false (nil for incr/decr) and close the connection after the request without retrying or down-marking, with the same log and trace diagnostics as reads. Write spans now carry the request opaque and correlation failure attributes.

Deletes, quiet/multi blocks, and multi-key writes are unchanged.
---
 CHANGELOG.md                                  |   5 +
 README.md                                     |  21 +-
 lib/dalli/client.rb                           |   6 +-
 lib/dalli/middlewares.rb                      |   2 +-
 lib/dalli/opentelemetry_middleware.rb         |  53 ++--
 lib/dalli/protocol/meta.rb                    |  45 ++--
 lib/dalli/protocol/meta/request_formatter.rb  |   6 +-
 lib/dalli/protocol/meta/response_processor.rb |   9 +-
 test/integration/test_opaque_correlation.rb   | 246 +++++++++++++++++-
 test/protocol/meta/test_request_formatter.rb  |  16 ++
 test/protocol/meta/test_response_processor.rb |  43 +++
 test/test_opentelemetry_middleware.rb         |  86 +++---
 12 files changed, 423 insertions(+), 115 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b12b60e0..08f78a24 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/README.md b/README.md
index a4920919..88d14280 100644
--- a/README.md
+++ b/README.md
@@ -27,14 +27,16 @@ 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:
 
@@ -42,13 +44,14 @@ For opt-in, per-request opaque correlation, also set `opaque_correlation_request
 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
 
diff --git a/lib/dalli/client.rb b/lib/dalli/client.rb
index 9c3e3d11..948eb64e 100644
--- a/lib/dalli/client.rb
+++ b/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
diff --git a/lib/dalli/middlewares.rb b/lib/dalli/middlewares.rb
index adbf61d9..d547128a 100644
--- a/lib/dalli/middlewares.rb
+++ b/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
 
diff --git a/lib/dalli/opentelemetry_middleware.rb b/lib/dalli/opentelemetry_middleware.rb
index 12d04cba..8aa2282a 100644
--- a/lib/dalli/opentelemetry_middleware.rb
+++ b/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
diff --git a/lib/dalli/protocol/meta.rb b/lib/dalli/protocol/meta.rb
index f0379f5c..72e85075 100644
--- a/lib/dalli/protocol/meta.rb
+++ b/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
 
diff --git a/lib/dalli/protocol/meta/request_formatter.rb b/lib/dalli/protocol/meta/request_formatter.rb
index 8e63bcc0..e9a6c9d7 100644
--- a/lib/dalli/protocol/meta/request_formatter.rb
+++ b/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
diff --git a/lib/dalli/protocol/meta/response_processor.rb b/lib/dalli/protocol/meta/response_processor.rb
index 9c266326..d7e4aa55 100644
--- a/lib/dalli/protocol/meta/response_processor.rb
+++ b/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
 
diff --git a/test/integration/test_opaque_correlation.rb b/test/integration/test_opaque_correlation.rb
index 710fae29..90d05ffb 100644
--- a/test/integration/test_opaque_correlation.rb
+++ b/test/integration/test_opaque_correlation.rb
@@ -4,10 +4,11 @@
 
 # TCP fixture for queued responses, incomplete bodies, and connection reuse.
 class OpaqueCorrelationServer
-  def initialize(&get_response)
+  def initialize(write_response: nil, &get_response)
     @listener = TCPServer.new('127.0.0.1', 0)
     @requests = Queue.new
     @get_response = get_response
+    @write_response = write_response || method(:echo_write_response)
     @thread = Thread.new { run }
   end
 
@@ -41,14 +42,23 @@ def response_for(command, line, connection_id)
     case command
     when 'version' then "VERSION 1.6.41-fake\r\n"
     when 'mg' then @get_response.call(line, connection_id)
-    when 'ma' then "VA 1\r\n2\r\n"
+    when 'ms', 'ma' then @write_response.call(line, connection_id)
     else raise "Unexpected command: #{line.inspect}"
     end
   end
 
+  # Like memcached, echo the request's O token and return a CAS only when asked for one.
+  def echo_write_response(line, _connection_id)
+    flags = line.split
+    opaque = flags.drop(2).find { |flag| flag.start_with?('O') }
+    suffix = opaque ? " #{opaque}" : ''
+    flags.first == 'ma' ? "VA 1#{suffix}\r\n2\r\n" : "HD#{' c7' if flags.include?('c')}#{suffix}\r\n"
+  end
+
   def serve(connection, connection_id)
     while (line = connection.gets("\r\n"))
       command = line.split.first
+      connection.read(line.split[2].to_i + 2) if command == 'ms'
       @requests << [connection_id, command] unless command == 'version'
       response = response_for(command, line, connection_id)
       break if response.nil?
@@ -74,12 +84,21 @@ def retrieve_req(operation, tags = {})
       yield attributes
     end
   end
+
+  def storage_req(operation, tags = {})
+    (@metric_tags ||= []).push(tags.dup, tags)
+    super do |attributes|
+      attributes = attributes.dup if attributes.frozen?
+      (@metric_attributes ||= []) << attributes
+      yield attributes
+    end
+  end
 end
 
-describe 'single-get opaque correlation' do
-  def with_opaque_server(response, correlate_with_opaques: true, **options)
+describe 'single-key opaque correlation' do
+  def with_opaque_server(response, correlate_with_opaques: true, write_response: nil, **options)
     options[:correlate_with_opaques] = correlate_with_opaques unless correlate_with_opaques.nil?
-    server = OpaqueCorrelationServer.new(&response)
+    server = OpaqueCorrelationServer.new(write_response: write_response, &response)
     client = Dalli::Client.new(server.address, raw: true, socket_timeout: 0.5,
                                                socket_failure_delay: nil, **options)
     # Bound regressions that accidentally drain a rejected body or retry.
@@ -175,6 +194,14 @@ def with_rejected_get(client, &)
     end
   end
 
+  # One public operation per write path: arguments before req_options, wire command, echoed result, rejected result.
+  write_operations = [
+    [:set, ['wanted', 'value', nil], 'ms', 7, false],
+    [:append, %w[wanted value], 'ms', true, false],
+    [:prepend, %w[wanted value], 'ms', true, false],
+    [:incr, ['counter', 1, nil, nil], 'ma', 2, nil]
+  ]
+
   modes = [nil, false, true]
   request_modes = [:omitted, nil, false, true]
   modes.product(modes, request_modes).each do |client_mode, request_only, request_mode|
@@ -206,6 +233,33 @@ def with_rejected_get(client, &)
         end
       end
     end
+
+    it "gates rollout writes with #{label}" do
+      wire = []
+      write_response = lambda do |line, _connection_id|
+        flags = line.split
+        opaque = flags.drop(2).find { |flag| flag.start_with?('O') }
+        wire << opaque
+        flags.first == 'ma' ? "VA 1 #{opaque || 'Oforeign'}\r\n2\r\n" : "HD c7 #{opaque || 'Oforeign'}\r\n"
+      end
+      options = request_mode == :omitted ? nil : { correlate_with_opaques: request_mode }.freeze
+      inherit = request_mode.nil? || request_mode == :omitted
+      enabled = client_mode == true && (inherit ? request_only != true : request_mode)
+      settings = { correlate_with_opaques: client_mode, opaque_correlation_request_only: request_only }
+
+      with_opaque_server(nil, write_response: write_response, **settings) do |client, server|
+        write_operations.each do |operation, args, _command, accepted, _rejected|
+          assert_equal accepted, client.public_send(operation, *args, options)
+        end
+
+        assert_equal(write_operations.map { |_operation, _args, command| [1, command] }, server.requests)
+        if enabled
+          wire.each { |opaque| assert_match(/\AO[A-Za-z0-9_-]{4}\z/, opaque) }
+        else
+          assert_equal Array.new(write_operations.size), wire
+        end
+      end
+    end
   end
 
   it 'uses the eagerly initialized PRNG across off/on/off reads without leaking trace metadata' do
@@ -355,25 +409,28 @@ def with_rejected_get(client, &)
   end
 
   [false, true].each do |enabled|
-    it "forwards request correlation=#{enabled} through fetch and cas!" do
+    it "forwards request correlation=#{enabled} through fetch and cas! reads and writes" do
       settings = { correlate_with_opaques: true, opaque_correlation_request_only: enabled, cache_nils: true }
       memcached(21_454, '', settings) do |client|
         client.set('rollout-compound', 'original')
         manager = client.send(:ring).servers.first.instance_variable_get(:@connection_manager)
         write = manager.method(:write)
-        reads = []
+        requests = []
         capture = lambda do |bytes|
-          reads << bytes if bytes.start_with?('mg ')
+          requests << bytes if bytes.start_with?('mg ', 'ms ')
           write.call(bytes)
         end
         manager.stub(:write, capture) do
           assert_equal 'original', client.fetch('rollout-compound', nil, correlate_with_opaques: enabled)
+          assert_equal 'filled', client.fetch('rollout-fill', nil, correlate_with_opaques: enabled) { 'filled' }
           result = client.cas!('rollout-compound', 30, correlate_with_opaques: enabled) { 'updated' }
 
           assert op_cas_succeeds(result)
         end
-        assert_equal 2, reads.size
-        assert_equal([enabled ? 1 : 0] * 2, reads.map { |line| line.split.drop(2).grep(/\AO/).size })
+
+        assert_equal(%w[mg mg ms mg ms], requests.map { |line| line[0, 2] })
+        assert_equal([enabled ? 1 : 0] * 5, requests.map { |line| line.split.drop(2).grep(/\AO/).size })
+        assert_equal 'filled', client.get('rollout-fill')
         assert_equal 'updated', client.get('rollout-compound')
       end
     end
@@ -787,4 +844,173 @@ def with_rejected_get(client, &)
       assert_equal [[1, 'mg'], [1, 'ma']], server.requests
     end
   end
+
+  [' Oobsolete', ''].each do |stale_flag|
+    write_operations.each do |operation, args, command, _accepted, rejected|
+      it "rejects a #{stale_flag.empty? ? 'bare' : 'stale'} #{operation} reply and recycles the connection" do
+        write_response = lambda do |line, _connection_id|
+          opaque = line.split.drop(2).find { |flag| flag.start_with?('O') }
+          [stale_flag, " #{opaque}"].map do |suffix|
+            command == 'ma' ? "VA 1#{suffix}\r\n9\r\n" : "HD c9#{suffix}\r\n"
+          end.join
+        end
+
+        with_opaque_server(nil, write_response: write_response, socket_max_failures: 1) do |client, server|
+          3.times do
+            result = client.public_send(operation, *args)
+
+            assert_operation_miss(result, operation, rejected)
+            assert_disconnected(client)
+          end
+
+          assert_equal [[1, command], [2, command], [3, command]], server.requests
+        end
+      end
+    end
+  end
+
+  it 'rejects a stale incr header without waiting for its counter line' do
+    write_response = lambda do |line, connection_id|
+      opaque = line.split.drop(2).find { |flag| flag.start_with?('O') }
+      connection_id == 1 ? "VA 1 Oobsolete\r\n" : "VA 1 #{opaque}\r\n2\r\n"
+    end
+
+    with_opaque_server(nil, write_response: write_response) do |client, server|
+      assert_nil client.incr('counter')
+      assert_disconnected(client)
+      assert_equal 2, client.incr('counter')
+      assert_equal [[1, 'ma'], [2, 'ma']], server.requests
+    end
+  end
+
+  it 'keeps real memcached connections open for every correlated write reply code' do
+    memcached_persistent(21_345, '', { raw: true, correlate_with_opaques: true }) do |client|
+      client.delete('opaque-write')
+      client.delete('opaque-counter')
+      server = client.send(:ring).servers.first
+      manager = server.instance_variable_get(:@connection_manager)
+      socket = server.sock
+      write = manager.method(:write)
+      requests = []
+      capture = lambda do |bytes|
+        requests << bytes if bytes.start_with?('ms ', 'ma ')
+        write.call(bytes)
+      end
+
+      manager.stub(:write, capture) do
+        refute client.replace('opaque-write', 'value') # NS
+        refute client.append('opaque-write', 'value') # NS
+        refute client.set_cas('opaque-write', 'value', 1) # NF
+        cas = client.set('opaque-write', 'value') # HD
+
+        assert op_addset_succeeds(cas)
+        refute client.add('opaque-write', 'value') # NS
+        refute client.set_cas('opaque-write', 'value', cas + 1) # EX
+        assert client.append('opaque-write', '>') # HD
+        assert client.prepend('opaque-write', '<') # HD
+        assert_nil client.incr('opaque-counter', 1) # NF
+        assert_equal 5, client.incr('opaque-counter', 1, nil, 5) # VA
+        assert_equal 3, client.decr('opaque-counter', 2) # VA
+      end
+
+      assert_same socket, server.sock
+      assert_equal '<value>', client.get('opaque-write')
+      assert_equal 11, requests.size
+      requests.each { |line| assert_equal 1, line.split.drop(2).grep(/\AO/).size, line }
+    end
+  end
+
+  it 'leaves deletes, quiet blocks, and multi-key writes uncorrelated' do
+    memcached_persistent(21_345, '', { raw: true, correlate_with_opaques: true }) do |client|
+      manager = client.send(:ring).servers.first.instance_variable_get(:@connection_manager)
+      write = manager.method(:write)
+      commands = []
+      capture = lambda do |bytes|
+        commands.concat(bytes.lines.grep(/\Am[sad] /))
+        write.call(bytes)
+      end
+
+      manager.stub(:write, capture) do
+        assert op_addset_succeeds(client.set('uncorrelated-control', 'value'))
+        client.set_multi({ 'uncorrelated-a' => 'a', 'uncorrelated-b' => 'b' }, 0)
+        client.quiet do
+          client.set('uncorrelated-c', 'c')
+          client.append('uncorrelated-c', 'c')
+          client.incr('uncorrelated-counter', 1, nil, 0)
+          client.delete('uncorrelated-a')
+        end
+
+        assert client.delete('uncorrelated-b')
+        refute client.delete_cas('uncorrelated-c', 1)
+        assert_equal 1, client.delete_multi(%w[uncorrelated-c uncorrelated-missing])
+      end
+      correlated = commands.select { |line| line.split.drop(2).grep(/\AO/).any? }
+
+      assert_equal 11, commands.size
+      assert_equal(['ms uncorrelated-control'], correlated.map { |line| line.split.first(2).join(' ') })
+    end
+  end
+
+  write_spans = { set: 'memcached.write', append: 'memcached.append', prepend: 'memcached.prepend',
+                  incr: 'memcached.incr' }
+  write_operations.each do |operation, args, command, accepted, rejected|
+    it "traces and logs #{operation} correlation without failing the span" do
+      OTEL_EXPORTER.reset
+      log = StringIO.new
+      logger = Logger.new(log)
+      logger.level = Logger::WARN
+      sent = []
+      write_response = lambda do |line, _connection_id|
+        opaque = line.split.drop(2).find { |flag| flag.start_with?('O') }
+        sent << opaque.delete_prefix('O')
+        reply = sent.size == 1 ? opaque : 'Oobsolete'
+        command == 'ma' ? "VA 1 #{reply}\r\n2\r\n" : "HD c7 #{reply}\r\n"
+      end
+      options = { write_response: write_response, middlewares: [Dalli::OpentelemetryMiddleware] }
+
+      Dalli.stub(:logger, logger) do
+        with_opaque_server(nil, **options) do |client, _server|
+          assert_equal accepted, client.public_send(operation, *args)
+          result = client.public_send(operation, *args)
+
+          assert_operation_miss(result, operation, rejected)
+          assert_disconnected(client)
+        end
+      end
+      success, failure = OTEL_EXPORTER.finished_spans.select { |span| span.name == write_spans.fetch(operation) }
+      code = command == 'ma' ? 'VA' : 'HD'
+
+      assert_equal(sent, [success, failure].map { |span| span.attributes['request_opaque'] })
+      refute success.attributes.key?('correlation_mismatch')
+      assert_equal 1, failure.attributes['correlation_mismatch']
+      assert_equal 'mismatch', failure.attributes['correlation_failure_reason']
+      assert_equal code, failure.attributes['response_code']
+      assert_equal 'obsolete', failure.attributes['received_opaque']
+      refute_equal OpenTelemetry::Trace::Status::ERROR, failure.status.code
+      assert_equal 1, log.string.scan('event=dalli.response_correlation_mismatch').size
+      assert_includes log.string, "response_code=#{code} reason=mismatch"
+    end
+  end
+
+  [false, true].each do |with_tracing|
+    it "keeps opaque metadata out of write metric inputs, tracing: #{with_tracing}" do
+      write_response = lambda do |line, connection_id|
+        opaque = connection_id == 1 ? 'Owrong' : line.split.drop(2).find { |flag| flag.start_with?('O') }
+        line.start_with?('ma ') ? "VA 1 #{opaque}\r\n2\r\n" : "HD c7 #{opaque}\r\n"
+      end
+      middlewares = [CorrelationMetricCapture]
+      middlewares.unshift(Dalli::OpentelemetryMiddleware) if with_tracing
+
+      with_opaque_server(nil, write_response: write_response, middlewares: middlewares) do |client, _server|
+        results = write_operations.map { |operation, args| client.public_send(operation, *args) }
+        stack = client.send(:ring).servers.first.instance_variable_get(:@middlewares_stack)
+        allowed_tags = %w[keys value_size ttl delta initial db.system]
+
+        assert_equal [false, true, true, 2], results
+        refute_empty stack.metric_tags
+        stack.metric_tags.each { |tags| assert_empty tags.keys - allowed_tags }
+        stack.metric_attributes.each { |attributes| assert_empty attributes }
+      end
+    end
+  end
 end
diff --git a/test/protocol/meta/test_request_formatter.rb b/test/protocol/meta/test_request_formatter.rb
index 351d5910..8c7fa54e 100644
--- a/test/protocol/meta/test_request_formatter.rb
+++ b/test/protocol/meta/test_request_formatter.rb
@@ -107,6 +107,14 @@
     let(:cas) { rand(500..999) }
     let(:ttl) { rand(500..999) }
 
+    it 'adds the supplied correlation token alongside CAS, TTL, and routing tokens' do
+      assert_equal "ms #{key} #{val.bytesize} c F#{bitflags} C#{cas} T#{ttl} MR Otoken Proute Lhint\r\n",
+                   Dalli::Protocol::Meta::RequestFormatter.meta_set(
+                     key: key, value: val, bitflags: bitflags, cas: cas, ttl: ttl, mode: :replace,
+                     opaque: 'token', p_token: 'route', l_token: 'hint'
+                   )
+    end
+
     it 'returns the default (treat as a set, no CAS check) when just passed key, datalen, and bitflags' do
       assert_equal "ms #{key} #{val.bytesize} c F#{bitflags} MS\r\n",
                    Dalli::Protocol::Meta::RequestFormatter.meta_set(key: key, value: val, bitflags: bitflags)
@@ -300,6 +308,14 @@
     let(:cas) { rand(500..999) }
     let(:ttl) { rand(500..999) }
 
+    it 'adds the supplied correlation token alongside routing tokens' do
+      assert_equal "ma #{key} v D#{delta} J#{initial} N#{ttl} MD Otoken Proute\r\n",
+                   Dalli::Protocol::Meta::RequestFormatter.meta_arithmetic(
+                     key: key, delta: delta, initial: initial, ttl: ttl, incr: false,
+                     opaque: 'token', p_token: 'route'
+                   )
+    end
+
     it 'returns the expected string with the default N flag when passed non-nil key, delta, and initial' do
       assert_equal "ma #{key} v D#{delta} J#{initial} N0 MI\r\n",
                    Dalli::Protocol::Meta::RequestFormatter.meta_arithmetic(key: key, delta: delta, initial: initial)
diff --git a/test/protocol/meta/test_response_processor.rb b/test/protocol/meta/test_response_processor.rb
index 818ed7c6..bda0b1ca 100644
--- a/test/protocol/meta/test_response_processor.rb
+++ b/test/protocol/meta/test_response_processor.rb
@@ -71,6 +71,49 @@ def processor_for(io, marshaller = nil)
     end
   end
 
+  write_parsers = {
+    meta_set_with_cas: [%w[HD NS EX NF], false],
+    meta_set_append_prepend: [%w[HD NS EX NF], false],
+    decr_incr: [%w[VA NF NS EX], nil]
+  }
+  [' Owrong-token', ' O', ''].each do |opaque_flag|
+    write_parsers.each do |method_name, (codes, rejected)|
+      codes.each do |code|
+        it "rejects #{method_name} #{code} replies with opaque #{opaque_flag.inspect} without reading a body" do
+          body = code == 'VA' ? "12\r\n" : ''
+          io = ResponseProcessorTestIO.new("#{code}#{' 2' if code == 'VA'}#{opaque_flag}\r\n", body)
+
+          result = processor_for(io).public_send(method_name, expected_opaque: 'expected-token')
+          received = opaque_flag.empty? ? nil : opaque_flag.strip[1..]
+
+          rejected.nil? ? assert_nil(result) : assert_equal(rejected, result)
+
+          assert_equal [['expected-token', received, code]], correlation_failures
+          assert_equal body, io.read_line.to_s, 'an uncorrelated counter line must not be consumed'
+        end
+      end
+    end
+  end
+
+  {
+    meta_set_with_cas: { 'HD c7' => 7, 'NS' => false, 'EX' => false, 'NF' => false },
+    meta_set_append_prepend: { 'HD' => true, 'NS' => false, 'EX' => false, 'NF' => false },
+    decr_incr: { 'VA 2' => 12, 'NF' => nil, 'NS' => false, 'EX' => false }
+  }.each do |method_name, replies|
+    replies.each do |header, expected|
+      it "parses #{method_name} #{header.split.first} replies carrying the expected opaque" do
+        io = ResponseProcessorTestIO.new("#{header} Oexpected-token\r\n", header.start_with?('VA') ? "12\r\n" : '')
+
+        result = processor_for(io).public_send(method_name, expected_opaque: 'expected-token')
+
+        expected.nil? ? assert_nil(result) : assert_equal(expected, result)
+
+        assert_empty correlation_failures
+        assert_nil io.read_line
+      end
+    end
+  end
+
   it 'extracts opaque strings and returns nil only when the flag is absent' do
     processor = processor_for(nil)
 
diff --git a/test/test_opentelemetry_middleware.rb b/test/test_opentelemetry_middleware.rb
index d1213c82..738a6166 100644
--- a/test/test_opentelemetry_middleware.rb
+++ b/test/test_opentelemetry_middleware.rb
@@ -4,53 +4,55 @@
 require 'dalli/opentelemetry_middleware'
 
 describe 'OpenTelemetry middleware' do
-  it 'keeps opaque metadata private and attaches it to the owning single-get span' do
-    OTEL_EXPORTER.reset
-    middleware = Dalli::Middlewares.new.extend(Dalli::OpentelemetryMiddleware)
-    tags = { 'keys' => 'key' }
-    shared_attributes = nil
-    middleware.retrieve_req('retrieval', tags) do |attributes|
-      shared_attributes = attributes
-      middleware.record_request_opaque('expected123')
-      Dalli::OpentelemetryMiddleware::TRACER.in_span('child') do
-        middleware.correlation_failure('correlation_mismatch' => 1, 'received_opaque' => 'received123')
+  %i[retrieve_req storage_req].each do |hook|
+    it "keeps opaque metadata private and attaches it to the owning #{hook} span" do
+      OTEL_EXPORTER.reset
+      middleware = Dalli::Middlewares.new.extend(Dalli::OpentelemetryMiddleware)
+      tags = { 'keys' => 'key' }
+      shared_attributes = nil
+      middleware.public_send(hook, 'request', tags) do |attributes|
+        shared_attributes = attributes
+        middleware.record_request_opaque('expected123')
+        Dalli::OpentelemetryMiddleware::TRACER.in_span('child') do
+          middleware.correlation_failure('correlation_mismatch' => 1, 'received_opaque' => 'received123')
+        end
+        attributes['miss_count'] = 1
       end
-      attributes['miss_count'] = 1
+      request = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'request' }
+      child = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'child' }
+
+      assert_equal({ 'keys' => 'key', 'db.system' => 'memcached' }, tags)
+      assert_equal({ 'miss_count' => 1 }, shared_attributes)
+      assert_equal 'expected123', request.attributes['request_opaque']
+      assert_equal 'received123', request.attributes['received_opaque']
+      assert_equal 1, request.attributes['correlation_mismatch']
+      refute child.attributes.key?('correlation_mismatch')
     end
-    retrieval = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'retrieval' }
-    child = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'child' }
-
-    assert_equal({ 'keys' => 'key', 'db.system' => 'memcached' }, tags)
-    assert_equal({ 'miss_count' => 1 }, shared_attributes)
-    assert_equal 'expected123', retrieval.attributes['request_opaque']
-    assert_equal 'received123', retrieval.attributes['received_opaque']
-    assert_equal 1, retrieval.attributes['correlation_mismatch']
-    refute child.attributes.key?('correlation_mismatch')
-  end
 
-  it 'does not apply trace metadata outside its retrieval scope, including after an interruption' do
-    OTEL_EXPORTER.reset
-    middleware = Dalli::Middlewares.new.extend(Dalli::OpentelemetryMiddleware)
-    cancellation = Exception.new('cancelled retrieval')
-    error = assert_raises(Exception) do
-      middleware.retrieve_req('interrupted') do |attributes|
-        attributes['partial_count'] = 1
-        middleware.record_request_opaque('expected123')
-        raise cancellation
+    it "does not apply trace metadata outside its #{hook} scope, including after an interruption" do
+      OTEL_EXPORTER.reset
+      middleware = Dalli::Middlewares.new.extend(Dalli::OpentelemetryMiddleware)
+      cancellation = Exception.new('cancelled request')
+      error = assert_raises(Exception) do
+        middleware.public_send(hook, 'interrupted') do |attributes|
+          attributes['partial_count'] = 1
+          middleware.record_request_opaque('expected123')
+          raise cancellation
+        end
       end
+      Dalli::OpentelemetryMiddleware::TRACER.in_span('unrelated') do
+        middleware.record_request_opaque('ignored')
+        middleware.correlation_failure('correlation_mismatch' => 1)
+      end
+      interrupted = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'interrupted' }
+      unrelated = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'unrelated' }
+
+      assert_same cancellation, error
+      refute interrupted.attributes.key?('request_opaque')
+      refute interrupted.attributes.key?('partial_count')
+      refute unrelated.attributes.key?('request_opaque')
+      refute unrelated.attributes.key?('correlation_mismatch')
     end
-    Dalli::OpentelemetryMiddleware::TRACER.in_span('unrelated') do
-      middleware.record_request_opaque('ignored')
-      middleware.correlation_failure('correlation_mismatch' => 1)
-    end
-    interrupted = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'interrupted' }
-    unrelated = OTEL_EXPORTER.finished_spans.find { |span| span.name == 'unrelated' }
-
-    assert_same cancellation, error
-    refute interrupted.attributes.key?('request_opaque')
-    refute interrupted.attributes.key?('partial_count')
-    refute unrelated.attributes.key?('request_opaque')
-    refute unrelated.attributes.key?('correlation_mismatch')
   end
 
   it 'restores private trace metadata after nested instrumentation' do