diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml
index af05ee0a..649174c9 100644
--- a/.github/workflows/benchmarks.yml
+++ b/.github/workflows/benchmarks.yml
@@ -8,10 +8,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v6
-    - name: Install Memcached 1.6.23
+    - name: Install Memcached 1.6.41
       working-directory: scripts
       env:
-        MEMCACHED_VERSION: 1.6.23
+        MEMCACHED_VERSION: 1.6.41
       run: |
         chmod +x ./install_memcached.sh
         ./install_memcached.sh
diff --git a/.github/workflows/profile.yml b/.github/workflows/profile.yml
index 8601466d..50d677f1 100644
--- a/.github/workflows/profile.yml
+++ b/.github/workflows/profile.yml
@@ -8,10 +8,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v6
-    - name: Install Memcached 1.6.23
+    - name: Install Memcached 1.6.41
       working-directory: scripts
       env:
-        MEMCACHED_VERSION: 1.6.23
+        MEMCACHED_VERSION: 1.6.41
       run: |
         chmod +x ./install_memcached.sh
         ./install_memcached.sh
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 637c503a..62c9bb90 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -14,7 +14,7 @@ jobs:
           - '3.4'
           - '3.3'
           - '3.2'
-        memcached-version: ['1.6.23']
+        memcached-version: ['1.6.41']
 
     steps:
     - uses: actions/checkout@v6
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2f3e5638..ae9afa49 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,18 @@ Dalli Changelog
 Unreleased
 ==========
 
+- Add tombstone (mark-stale) support to `Client#delete` / `delete_cas` /
+  `delete_multi` via new `:invalidate`, `:tombstone_ttl`, `:drop_value`
+  request-option keys (corresponding to meta-protocol `I`, `T`, `x` flags
+  on `md`). A tombstoned item lives briefly in a "stale" window so
+  concurrent readers can tell a racing repopulate apart from a true miss
+  — useful for high-concurrency cache invalidation. (drinkbeer)
+- Add `Client#get_with_status` returning a `Dalli::CacheResult` value
+  object with `value` / `stale?` / `miss?` / `hit?` predicates. Unlike
+  `#get`, it always returns a result (never nil) and surfaces the
+  meta-protocol `X` (stale) response flag, letting callers distinguish
+  a tombstone window from a true miss without changing the return shape
+  of `get`. (drinkbeer)
 - Fix cannot read response data included terminator `\r\n` when use meta protocol (matsubara0507)
 - Remove binary protocol support (grcooper)
 - Add support for `raw` client option (nherson)
diff --git a/lib/dalli.rb b/lib/dalli.rb
index d3079b97..01d3c3b9 100644
--- a/lib/dalli.rb
+++ b/lib/dalli.rb
@@ -67,6 +67,7 @@ def self.register(middleware)
 require_relative 'dalli/version'
 require_relative 'dalli/middlewares'
 
+require_relative 'dalli/cache_result'
 require_relative 'dalli/compressor'
 require_relative 'dalli/client'
 require_relative 'dalli/key_manager'
diff --git a/lib/dalli/cache_result.rb b/lib/dalli/cache_result.rb
new file mode 100644
index 00000000..277a19b7
--- /dev/null
+++ b/lib/dalli/cache_result.rb
@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+module Dalli
+  # Result of a stale-aware read (Client#get_with_status). Always returned —
+  # callers should branch on the predicate methods, not on nil-ness, since a
+  # tombstoned item has stale? == true with a (possibly empty) value, while
+  # a real cache miss has miss? == true.
+  class CacheResult
+    attr_reader :value
+
+    def initialize(value:, stale: false, miss: false)
+      @value = value
+      @stale = stale
+      @miss = miss
+      freeze
+    end
+
+    def stale?
+      @stale
+    end
+
+    def miss?
+      @miss
+    end
+
+    def hit?
+      !@miss
+    end
+  end
+end
diff --git a/lib/dalli/client.rb b/lib/dalli/client.rb
index 26daa033..76163bc4 100644
--- a/lib/dalli/client.rb
+++ b/lib/dalli/client.rb
@@ -93,6 +93,21 @@ def get_cas(key, req_options = nil)
       yield value, cas
     end
 
+    ##
+    # Read a key and return a `Dalli::CacheResult` with stale-awareness.
+    #
+    # Unlike `#get`, this always returns a `CacheResult` (never nil). Callers
+    # branch on `result.stale?` / `result.miss?` / `result.hit?` rather than
+    # nil-ness, since a tombstoned item has `stale? == true` with a (possibly
+    # empty) value, while a real cache miss has `miss? == true`.
+    #
+    # Tombstones are produced via `#delete` with `invalidate: true`; `drop_value`
+    # can be combined to discard the previous value while keeping the stale marker.
+    # See `#delete` for details.
+    def get_with_status(key, req_options = nil)
+      perform(:get_with_status, key, req_options)
+    end
+
     ##
     # Fetch multiple keys efficiently.
     # If a block is given, yields key/value pairs one at a time.
@@ -129,6 +144,35 @@ def get_multi(*keys, **req_options)
       end
     end
 
+    ##
+    # Fetch multiple keys efficiently and return stale-aware `Dalli::CacheResult`
+    # objects for every requested key.
+    #
+    # Unlike `#get_multi`, the returned hash includes misses so callers can
+    # distinguish a true miss from a tombstoned/stale item for each key:
+    #   { 'key' => #<Dalli::CacheResult ...>, 'missing' => #<Dalli::CacheResult miss? ...> }
+    #
+    # If a block is given, yields key/result pairs one at a time for every
+    # requested key.
+    #
+    # See `get_multi` for documentation on the `req_options` trailing keyword
+    # arguments (e.g. `p_token:` / `l_token:`), including the kwargs-vs-positional caveat.
+    def get_multi_with_status(*keys, **req_options, &block)
+      keys.flatten!
+      keys.compact!
+
+      return {} if keys.empty?
+
+      req_options = nil if req_options.empty?
+      results = pipelined_getter.process_with_status(keys, req_options)
+
+      if block
+        results.each(&block)
+      else
+        results
+      end
+    end
+
     ##
     # Fetch multiple keys efficiently, including available metadata such as CAS.
     # If a block is given, yields key/data pairs one a time.  Data is an array:
@@ -273,10 +317,28 @@ def replace_cas(key, value, cas, ttl = nil, req_options = nil)
 
     # Delete a key/value pair, verifying existing CAS.
     # Returns true if succeeded, and falsy otherwise.
+    #
+    # `req_options` recognizes the same meta-delete keys as `#delete`:
+    # `:invalidate`, `:tombstone_ttl`, `:drop_value`.
     def delete_cas(key, cas = 0, req_options = nil)
       perform(:delete, key, cas, req_options)
     end
 
+    ##
+    # Delete a key.
+    #
+    # `req_options` may include memcached meta-delete options:
+    # - `:invalidate` (Boolean) — mark the item stale instead of removing it.
+    #   This is the tombstone marker: `#get_with_status` returns `stale?`, and
+    #   the existing value remains readable unless `:drop_value` is also set.
+    # - `:drop_value` (Boolean) — remove the item value but leave the item.
+    #   Alone this is not a tombstone: reads are a non-stale hit with an empty
+    #   string value.
+    # - `:invalidate` + `:drop_value` — leave a stale tombstone marker with an
+    #   empty value, so readers can distinguish it from a miss without retaining
+    #   the previous value.
+    # - `:tombstone_ttl` (Integer seconds) — how long the stale tombstone lives;
+    #   requires `:invalidate`. After this elapses, reads see `miss?`.
     def delete(key, req_options = nil)
       delete_cas(key, 0, req_options)
     end
@@ -285,9 +347,9 @@ def delete(key, req_options = nil)
     # Delete multiple keys efficiently in pipelined mode.
     # Returns the number of keys that were successfully deleted.
     #
-    # `req_options` is applied to every delete in the pipeline (e.g.
-    # `meta_flags: ['Proute=...']`). Best-effort; the same options apply to
-    # every key.
+    # `req_options` is applied to every delete in the pipeline. Recognized
+    # meta-delete keys (`:invalidate`, `:tombstone_ttl`, `:drop_value`) are
+    # applied uniformly to every key in the batch — see `#delete`.
     def delete_multi(keys, req_options = nil)
       return 0 if keys.empty?
 
diff --git a/lib/dalli/pipelined_getter.rb b/lib/dalli/pipelined_getter.rb
index c3c35b95..efa3a78e 100644
--- a/lib/dalli/pipelined_getter.rb
+++ b/lib/dalli/pipelined_getter.rb
@@ -16,6 +16,27 @@ def optimized_for_single_server(keys, req_options = nil)
       @key_manager.key_values_without_namespace(results)
     end
 
+    def process_with_status(keys, req_options = nil)
+      return {} if keys.empty?
+
+      @ring.lock do
+        groups = groups_for_keys(keys)
+        results = groups.each_with_object({}) do |(server, keys_for_server), hash|
+          hash.merge!(server.request(:read_multi_with_status_req, keys_for_server, req_options))
+        rescue RetryableNetworkError
+          raise
+        rescue DalliError => e
+          Dalli.logger.debug { e.inspect }
+          Dalli.logger.debug { "unable to get keys for server #{server.name}" }
+        end
+        @key_manager.key_values_without_namespace(results)
+      end
+    rescue RetryableNetworkError => e
+      Dalli.logger.debug { e.inspect }
+      Dalli.logger.debug { 'retrying pipelined get with status because of timeout' }
+      retry
+    end
+
     ##
     # Yields, one at a time, keys and their values+attributes.
     #
diff --git a/lib/dalli/protocol/base.rb b/lib/dalli/protocol/base.rb
index d9cec641..8bbeaa95 100644
--- a/lib/dalli/protocol/base.rb
+++ b/lib/dalli/protocol/base.rb
@@ -253,6 +253,27 @@ def routing_tokens?(opts)
         !blank_token?(opts[:p_token]) || !blank_token?(opts[:l_token])
       end
 
+      # Extracts meta-delete kwargs (`invalidate`, `tombstone_ttl`,
+      # `drop_value`) from a request-options Hash so they can be splatted
+      # into a RequestFormatter.meta_delete call. Returns `{}` when none
+      # of the keys are set, so the splat is a no-op for the common path.
+      # Tombstone TTL uses the same memcached expiration semantics as other
+      # TTL-bearing operations, so sanitize it before formatting the request.
+      # Validation (e.g. `tombstone_ttl` requiring `invalidate`) happens at
+      # the wire-formatter level, where it can ArgumentError uniformly.
+      def tombstone_kwargs(opts)
+        return {} unless opts.is_a?(Hash)
+
+        invalidate    = opts[:invalidate]
+        tombstone_ttl = opts[:tombstone_ttl]
+        drop_value    = opts[:drop_value]
+        return {} unless invalidate || tombstone_ttl || drop_value
+
+        tombstone_ttl = TtlSanitizer.sanitize(Integer(tombstone_ttl)) if tombstone_ttl
+
+        { invalidate: invalidate, tombstone_ttl: tombstone_ttl, drop_value: drop_value }.compact
+      end
+
       def blank_token?(value)
         value.nil? || (value.respond_to?(:empty?) && value.empty?)
       end
diff --git a/lib/dalli/protocol/meta.rb b/lib/dalli/protocol/meta.rb
index dc003f8c..92fefe7c 100644
--- a/lib/dalli/protocol/meta.rb
+++ b/lib/dalli/protocol/meta.rb
@@ -11,7 +11,7 @@ module Protocol
     # protocol.  Contains logic for managing connection state to the server (retries, etc),
     # formatting requests to the server, and unpacking responses.
     ##
-    class Meta < Base
+    class Meta < Base # rubocop:disable Metrics/ClassLength
       TERMINATOR = "\r\n"
       SUPPORTS_CAPACITY = Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('3.4.0')
 
@@ -70,14 +70,13 @@ def read_multi_req(keys, req_options = nil)
         # Pre-allocate the results hash with expected size
         results = SUPPORTS_CAPACITY ? Hash.new(nil, capacity: keys.size) : {}
         optimized_for_raw = @value_marshaller.raw_by_default
-        key_index = optimized_for_raw ? 2 : 3
 
         total_value_bytesize = 0
         @middlewares_stack.retrieve_req_pipeline('memcached.read_multi', { 'keys' => keys }) do |attributes|
           routing_suffix = RequestFormatter.routing_tokens(**routing_token_kwargs(req_options))
           post_get_req = optimized_for_raw ? "v k q#{routing_suffix}\r\n" : "v f k q#{routing_suffix}\r\n"
           keys.each do |key|
-            @connection_manager.write("mg #{key} #{post_get_req}")
+            write_read_multi_get(key, post_get_req)
           end
           @connection_manager.write("mn\r\n")
           @connection_manager.flush
@@ -92,7 +91,7 @@ def read_multi_req(keys, req_options = nil)
             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
-            key = tokens[key_index]&.byteslice(1..-1)
+            key = response_processor.key_from_tokens(tokens)
             next if key.nil?
 
             total_value_bytesize += value.bytesize
@@ -112,13 +111,83 @@ def read_multi_req(keys, req_options = nil)
       # rubocop:enable Metrics/PerceivedComplexity
       # rubocop:enable Metrics/MethodLength
 
+      # rubocop:disable Metrics/CyclomaticComplexity
+      # rubocop:disable Metrics/MethodLength
+      # rubocop:disable Metrics/PerceivedComplexity
+      def read_multi_with_status_req(keys, req_options = nil)
+        results = SUPPORTS_CAPACITY ? Hash.new(nil, capacity: keys.size) : {}
+        optimized_for_raw = @value_marshaller.raw_by_default
+
+        total_value_bytesize = 0
+        fresh_hit_count = 0
+        stale_count = 0
+        @middlewares_stack.retrieve_req_pipeline('memcached.read_multi_with_status', { 'keys' => keys }) do |attributes|
+          routing_suffix = RequestFormatter.routing_tokens(**routing_token_kwargs(req_options))
+          post_get_req = optimized_for_raw ? "v k q#{routing_suffix}\r\n" : "v f k q#{routing_suffix}\r\n"
+          keys.each do |key|
+            write_read_multi_get(key, post_get_req)
+          end
+          @connection_manager.write("mn\r\n")
+          @connection_manager.flush
+
+          terminator_length = TERMINATOR.length
+          while (line = @connection_manager.readline)
+            break if line == TERMINATOR || line[0, 2] == 'MN'
+            next unless line[0, 3] == 'VA '
+
+            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)
+            key = response_processor.key_from_tokens(tokens)
+            next if key.nil?
+
+            stale = response_processor.stale_from_tokens(tokens)
+            total_value_bytesize += value.bytesize
+            if stale
+              stale_count += 1
+            else
+              fresh_hit_count += 1
+            end
+            results[key] = ::Dalli::CacheResult.new(
+              value: @value_marshaller.retrieve(value, bitflags),
+              stale: stale
+            )
+          end
+
+          keys.each do |key|
+            results[key] = ::Dalli::CacheResult.new(value: nil, miss: true) unless results.key?(key)
+          end
+
+          unless attributes.frozen?
+            # Stale tombstones are CacheResult#hit? at the API layer, but
+            # count as non-fresh for hit-rate metrics.
+            attributes['value_bytesize'] = total_value_bytesize
+            attributes['hit_count'] = fresh_hit_count
+            attributes['miss_count'] = keys.size - fresh_hit_count
+            attributes['stale_count'] = stale_count
+          end
+        end
+
+        results
+      end
+      # rubocop:enable Metrics/CyclomaticComplexity
+      # rubocop:enable Metrics/MethodLength
+      # rubocop:enable Metrics/PerceivedComplexity
+
+      def write_read_multi_get(key, post_get_req)
+        encoded_key, base64 = KeyRegularizer.encode(key)
+        @connection_manager.write("mg #{encoded_key}#{' b' if base64} #{post_get_req}")
+      end
+
       def delete_multi_req(keys, req_options = nil)
         routing_kwargs = routing_token_kwargs(req_options)
+        tombstone_extras = tombstone_kwargs(req_options)
         @middlewares_stack.storage_req_pipeline('delete_multi', { 'keys' => keys }) do
           keys.each do |key|
             encoded_key, base64 = KeyRegularizer.encode(key)
             req = RequestFormatter.meta_delete(key: encoded_key, base64: base64, quiet: true,
-                                               **routing_kwargs)
+                                               **routing_kwargs, **tombstone_extras)
             write(req)
           end
           write_noop
@@ -181,6 +250,29 @@ def quiet_get_request(key, req_options = nil)
         end
       end
 
+      def get_with_status(key, options = nil)
+        encoded_key, base64 = KeyRegularizer.encode(key)
+        routing_kwargs = routing_token_kwargs(options)
+
+        @middlewares_stack.retrieve_req('memcached.get_with_status', { 'keys' => key }) do |attributes|
+          req = RequestFormatter.meta_get(key: encoded_key, value: true, base64: base64,
+                                          **routing_kwargs)
+          write(req)
+          @connection_manager.flush
+          result, raw_value_bytesize = response_processor.meta_get_with_status
+          unless attributes.frozen?
+            # Stale tombstones are CacheResult#hit? at the API layer, but
+            # count as non-fresh for hit-rate metrics.
+            fresh_hit = result.hit? && !result.stale?
+            attributes['value_bytesize'] = raw_value_bytesize
+            attributes['hit_count'] = fresh_hit ? 1 : 0
+            attributes['miss_count'] = fresh_hit ? 0 : 1
+            attributes['stale_count'] = result.stale? ? 1 : 0
+          end
+          result
+        end
+      end
+
       # rubocop:disable Metrics/AbcSize
       def gat(key, ttl, options = nil)
         ttl = TtlSanitizer.sanitize(ttl)
@@ -313,7 +405,8 @@ def delete(key, cas, options = nil)
         @middlewares_stack.storage_req('memcached.delete', { 'keys' => key, 'cas' => cas }) do
           req = RequestFormatter.meta_delete(key: encoded_key, cas: cas,
                                              base64: base64, quiet: quiet?,
-                                             **routing_token_kwargs(options))
+                                             **routing_token_kwargs(options),
+                                             **tombstone_kwargs(options))
           write(req)
           @connection_manager.flush
           response_processor.meta_delete unless quiet?
diff --git a/lib/dalli/protocol/meta/request_formatter.rb b/lib/dalli/protocol/meta/request_formatter.rb
index 1daabe0f..b51eab4b 100644
--- a/lib/dalli/protocol/meta/request_formatter.rb
+++ b/lib/dalli/protocol/meta/request_formatter.rb
@@ -42,12 +42,18 @@ def self.meta_set(key:, value:, bitflags: nil, cas: nil, ttl: nil, mode: :set, b
           cmd << TERMINATOR
         end
 
-        def self.meta_delete(key:, cas: nil, ttl: nil, base64: false, quiet: false, p_token: nil, l_token: nil)
+        def self.meta_delete(key:, cas: nil, ttl: nil, base64: false, quiet: false, p_token: nil, l_token: nil,
+                             invalidate: false, tombstone_ttl: nil, drop_value: false)
+          raise ArgumentError, 'tombstone_ttl requires invalidate: true' if tombstone_ttl && !invalidate
+
           cmd = "md #{key}"
           cmd << ' b' if base64
           cmd << cas_string(cas)
           cmd << " T#{ttl}" if ttl
           cmd << ' q' if quiet
+          cmd << ' I' if invalidate
+          cmd << " T#{Integer(tombstone_ttl)}" if tombstone_ttl
+          cmd << ' x' if drop_value
           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 1ae56ca3..ea1647de 100644
--- a/lib/dalli/protocol/meta/response_processor.rb
+++ b/lib/dalli/protocol/meta/response_processor.rb
@@ -67,6 +67,25 @@ def meta_get_without_value
           tokens.first == EN ? nil : true
         end
 
+        # Stale-aware get that returns a Dalli::CacheResult and the raw
+        # response body size so metrics can report wire bytes rather than
+        # calling #bytesize on the deserialized value.
+        # The value field may be empty when the tombstone was created with
+        # drop_value, which is intentional — callers branch on the
+        # predicates rather than nil-ness.
+        def meta_get_with_status
+          tokens = error_on_unexpected!([VA, EN, HD])
+          return [::Dalli::CacheResult.new(value: nil, miss: true), 0] if tokens.first == EN
+
+          if tokens.first == VA
+            raw_value = read_data(tokens[1].to_i)
+            value = @value_marshaller.retrieve(raw_value, bitflags_from_tokens(tokens))
+            [::Dalli::CacheResult.new(value: value, stale: stale_from_tokens(tokens)), raw_value.bytesize]
+          else
+            [::Dalli::CacheResult.new(value: nil, miss: true), 0]
+          end
+        end
+
         def meta_set_with_cas
           tokens = error_on_unexpected!([HD, NS, NF, EX])
           return false unless tokens.first == HD
@@ -218,6 +237,14 @@ def bitflags_from_tokens(tokens)
           value_from_tokens(tokens, 'f')&.to_i
         end
 
+        # Detects the `X` presence flag indicating the item has been marked
+        # stale via a prior `md key I`. Strict equality (Array#any?(pattern)
+        # uses `===`, which for Strings is `==`) avoids false positives if a
+        # future value-bearing flag is introduced beginning with `X`.
+        def stale_from_tokens(tokens)
+          tokens.any?('X')
+        end
+
         def cas_from_tokens(tokens)
           value_from_tokens(tokens, 'c')&.to_i
         end
diff --git a/test/integration/test_pipelined_get.rb b/test/integration/test_pipelined_get.rb
index 7386ac8b..0a0b3bff 100644
--- a/test/integration/test_pipelined_get.rb
+++ b/test/integration/test_pipelined_get.rb
@@ -153,32 +153,35 @@
     end
   end
 
-  it 'supports pipelined get with keys containing Unicode' do
+  it 'supports pipelined get with keys requiring base64 encoding' do
     memcached_persistent do |dc|
       dc.close
       dc.flush
 
-      keys_to_query = ['a', 'b', 'ƒ©åÍÎ']
+      unicode_key = 'ƒ©åÍÎ'
+      space_key = 'space key'
+      crlf_key = "crlf\r\nkey"
+      keys_to_query = ['a', 'b', unicode_key, space_key, crlf_key]
 
       resp = dc.get_multi(keys_to_query)
 
       assert_empty(resp)
 
       dc.set('a', 'foo')
-      dc.set('ƒ©åÍÎ', %w[a b c])
+      dc.set(unicode_key, %w[a b c])
+      dc.set(space_key, 'space')
+      dc.set(crlf_key, 'crlf')
 
       # Invocation without block
       resp = dc.get_multi(keys_to_query)
-      expected_resp = { 'a' => 'foo', Dalli::Protocol::Meta::KeyRegularizer.encode('ƒ©åÍÎ')[0] => %w[a b c] }
+      expected_resp = { 'a' => 'foo', unicode_key => %w[a b c], space_key => 'space', crlf_key => 'crlf' }
 
       assert_equal(expected_resp, resp)
 
       # Invocation with block
       dc.get_multi(keys_to_query) do |k, v|
-        encoded_key = Dalli::Protocol::Meta::KeyRegularizer.encode(k)[0]
-
-        assert(expected_resp.key?(encoded_key) && expected_resp[encoded_key] == v)
-        expected_resp.delete(encoded_key)
+        assert(expected_resp.key?(k) && expected_resp[k] == v)
+        expected_resp.delete(k)
       end
 
       assert_empty expected_resp
diff --git a/test/integration/test_routing_tokens_passthrough.rb b/test/integration/test_routing_tokens_passthrough.rb
index 0d8cc846..672d9425 100644
--- a/test/integration/test_routing_tokens_passthrough.rb
+++ b/test/integration/test_routing_tokens_passthrough.rb
@@ -233,6 +233,23 @@
     end
   end
 
+  describe 'get_multi_with_status' do
+    it 'applies routing tokens to every key in the status-aware pipeline' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('a', '1')
+        dc.set('b', '2')
+
+        results = dc.get_multi_with_status('a', 'b', 'missing', **ROUTING_OPTS)
+
+        assert_equal %w[a b missing], results.keys.sort
+        assert_equal '1', results['a'].value
+        assert_equal '2', results['b'].value
+        assert_predicate results['missing'], :miss?
+      end
+    end
+  end
+
   describe 'set_multi (pipelined)' do
     it 'applies routing tokens to every entry in the pipeline' do
       memcached_persistent do |dc|
diff --git a/test/integration/test_tombstone.rb b/test/integration/test_tombstone.rb
new file mode 100644
index 00000000..5205036f
--- /dev/null
+++ b/test/integration/test_tombstone.rb
@@ -0,0 +1,390 @@
+# frozen_string_literal: true
+
+require_relative '../helper'
+
+# Integration tests for memcached tombstone support: the meta-protocol's
+# `I` (mark stale), `T` (tombstone TTL on delete), and `x` (drop value)
+# flags on `md`, plus the `X` response flag on `mg` exposed via
+# Client#get_with_status.
+#
+# A tombstoned item lives briefly in a "stale" window where reads can tell
+# a racing repopulator from a true miss — caller checks result.stale? vs
+# result.miss?. Designed for high-concurrency invalidation scenarios where
+# a hard delete would let an in-flight request rewrite a stale value over
+# the deleted key.
+describe 'tombstone (mark-stale) support' do
+  describe 'Client#get_with_status return shape' do
+    it 'returns a Dalli::CacheResult with hit? on a normal hit' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk', 'val'))
+
+        result = dc.get_with_status('tk')
+
+        assert_kind_of Dalli::CacheResult, result
+        assert_equal 'val', result.value
+        assert_predicate result, :hit?
+        refute_predicate result, :miss?
+        refute_predicate result, :stale?
+      end
+    end
+
+    it 'returns miss? on a true miss (key never existed)' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        result = dc.get_with_status('absent')
+
+        assert_kind_of Dalli::CacheResult, result
+        assert_nil result.value
+        assert_predicate result, :miss?
+        refute_predicate result, :hit?
+        refute_predicate result, :stale?
+      end
+    end
+
+    it 'returns miss? after a regular (non-tombstone) delete' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk', 'val'))
+        dc.delete('tk')
+
+        result = dc.get_with_status('tk')
+
+        assert_predicate result, :miss?
+        refute_predicate result, :stale?
+      end
+    end
+
+    it 'is frozen so callers cannot mutate the returned object' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tk', 'val')
+
+        assert_predicate dc.get_with_status('tk'), :frozen?
+      end
+    end
+
+    # Pins the three-state contract: callers branch on miss?/stale?/hit?,
+    # never on value.nil? — because a legitimately cached nil is a hit.
+    it 'distinguishes a cached nil (hit?), an absent key (miss?), and a tombstone (stale?)' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk-nil', nil))
+        cached_nil = dc.get_with_status('tk-nil')
+
+        assert_predicate cached_nil, :hit?
+        refute_predicate cached_nil, :miss?
+        refute_predicate cached_nil, :stale?
+        assert_nil cached_nil.value
+
+        absent = dc.get_with_status('tk-absent')
+
+        assert_predicate absent, :miss?
+        refute_predicate absent, :hit?
+        refute_predicate absent, :stale?
+        assert_nil absent.value
+
+        assert op_addset_succeeds(dc.set('tk-tomb', 'val'))
+        dc.delete('tk-tomb', invalidate: true, tombstone_ttl: 30)
+        tombstoned = dc.get_with_status('tk-tomb')
+
+        assert_predicate tombstoned, :stale?
+        assert_predicate tombstoned, :hit?
+        refute_predicate tombstoned, :miss?
+      end
+    end
+  end
+
+  describe 'Client#get_multi_with_status return shape' do
+    it 'returns CacheResult values for normal hits, stale tombstones, dropped tombstones, and misses' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('multi-hit', 'hit-val'))
+        assert op_addset_succeeds(dc.set('multi-stale', 'stale-val'))
+        assert op_addset_succeeds(dc.set('multi-dropped', 'dropped-val'))
+        dc.delete('multi-stale', invalidate: true, tombstone_ttl: 30)
+        dc.delete('multi-dropped', invalidate: true, drop_value: true, tombstone_ttl: 30)
+
+        results = dc.get_multi_with_status('multi-hit', 'multi-stale', 'multi-dropped', 'multi-absent')
+
+        assert_equal %w[multi-absent multi-dropped multi-hit multi-stale], results.keys.sort
+
+        hit = results['multi-hit']
+
+        assert_kind_of Dalli::CacheResult, hit
+        assert_equal 'hit-val', hit.value
+        assert_predicate hit, :hit?
+        refute_predicate hit, :miss?
+        refute_predicate hit, :stale?
+
+        stale = results['multi-stale']
+
+        assert_kind_of Dalli::CacheResult, stale
+        assert_equal 'stale-val', stale.value
+        assert_predicate stale, :hit?
+        refute_predicate stale, :miss?
+        assert_predicate stale, :stale?
+
+        dropped = results['multi-dropped']
+
+        assert_kind_of Dalli::CacheResult, dropped
+        assert_equal '', dropped.value
+        assert_predicate dropped, :hit?
+        refute_predicate dropped, :miss?
+        assert_predicate dropped, :stale?
+
+        absent = results['multi-absent']
+
+        assert_kind_of Dalli::CacheResult, absent
+        assert_nil absent.value
+        assert_predicate absent, :miss?
+        refute_predicate absent, :hit?
+        refute_predicate absent, :stale?
+      end
+    end
+
+    it 'returns results under original keys when keys require base64 encoding' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        unicode_key = 'multi-ƒ©åÍÎ'
+        space_key = 'multi space'
+        crlf_key = "multi-crlf\r\nkey"
+        absent_key = "multi-absent\r\nkey"
+
+        assert op_addset_succeeds(dc.set(unicode_key, 'unicode-val'))
+        assert op_addset_succeeds(dc.set(space_key, 'space-val'))
+        assert op_addset_succeeds(dc.set(crlf_key, 'crlf-val'))
+        dc.delete(unicode_key, invalidate: true, tombstone_ttl: 30)
+
+        results = dc.get_multi_with_status(space_key, unicode_key, crlf_key, absent_key)
+
+        assert_equal [absent_key, crlf_key, space_key, unicode_key].sort, results.keys.sort
+        assert_equal 'space-val', results[space_key].value
+        assert_equal 'crlf-val', results[crlf_key].value
+        assert_equal 'unicode-val', results[unicode_key].value
+        assert_predicate results[space_key], :hit?
+        assert_predicate results[crlf_key], :hit?
+        assert_predicate results[unicode_key], :stale?
+        assert_predicate results[absent_key], :miss?
+      end
+    end
+
+    it 'yields a CacheResult for every requested key in block form' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('multi-block-hit', 'hit-val'))
+
+        seen = {}
+        dc.get_multi_with_status('multi-block-hit', 'multi-block-absent') do |key, result|
+          seen[key] = result
+        end
+
+        assert_equal %w[multi-block-absent multi-block-hit], seen.keys.sort
+        assert_equal 'hit-val', seen['multi-block-hit'].value
+        assert_predicate seen['multi-block-hit'], :hit?
+        assert_predicate seen['multi-block-absent'], :miss?
+      end
+    end
+  end
+
+  describe 'delete with drop_value only' do
+    it 'drops the value without marking the item stale' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        # Use a non-raw value so this also proves the zero-byte response from
+        # `md ... x` does not try to unmarshal the previously serialized value.
+        assert op_addset_succeeds(dc.set('tk-x-only', { payload: 'should-be-dropped' }))
+
+        assert dc.delete('tk-x-only', drop_value: true)
+        result = dc.get_with_status('tk-x-only')
+
+        assert_predicate result, :hit?
+        refute_predicate result, :stale?
+        refute_predicate result, :miss?
+        assert_equal '', result.value
+        assert_equal '', dc.get('tk-x-only')
+      end
+    end
+  end
+
+  describe 'delete with invalidate: true' do
+    it 'leaves the item readable but marked stale' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk', 'preserved-val'))
+
+        dc.delete('tk', invalidate: true)
+        result = dc.get_with_status('tk')
+
+        assert_predicate result, :stale?, 'expected X flag (stale) after invalidate'
+        assert_predicate result, :hit?, 'invalidate without drop_value should leave value readable'
+        refute_predicate result, :miss?
+        assert_equal 'preserved-val', result.value
+      end
+    end
+
+    it 'leaves an empty value when drop_value is also set' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk', 'should-be-dropped'))
+
+        dc.delete('tk', invalidate: true, drop_value: true)
+        result = dc.get_with_status('tk')
+
+        assert_predicate result, :stale?
+        refute_predicate result, :miss?
+        # Value is dropped — empty string, not the original
+        assert_equal '', result.value
+      end
+    end
+
+    it 'transitions from stale? to miss? after tombstone_ttl elapses' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk', 'val'))
+
+        dc.delete('tk', invalidate: true, tombstone_ttl: 1, drop_value: true)
+
+        # Within the tombstone window
+        assert_predicate dc.get_with_status('tk'), :stale?
+
+        # Past the tombstone window, the X flag should be gone
+        sleep 2
+        result = dc.get_with_status('tk')
+
+        assert_predicate result, :miss?, 'tombstone should have expired into a true miss'
+        refute_predicate result, :stale?
+      end
+    end
+
+    it 'sanitizes long tombstone_ttl intervals before sending to memcached' do
+      memcached_persistent do |dc|
+        dc.flush
+
+        assert op_addset_succeeds(dc.set('tk-long-ttl', 'val'))
+
+        long_ttl = Dalli::Protocol::TtlSanitizer::MAX_ACCEPTABLE_EXPIRATION_INTERVAL + 1
+        dc.delete('tk-long-ttl', invalidate: true, tombstone_ttl: long_ttl)
+
+        assert_predicate dc.get_with_status('tk-long-ttl'), :stale?
+      end
+    end
+
+    it 'does not emit a tombstone for a plain delete' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tk', 'val')
+
+        dc.delete('tk') # no kwargs — regular hard delete
+        result = dc.get_with_status('tk')
+
+        assert_predicate result, :miss?
+        refute_predicate result, :stale?
+      end
+    end
+
+    it 'is reachable via delete_cas with explicit cas' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tk', 'val')
+        cas = dc.get_cas('tk').last
+
+        dc.delete_cas('tk', cas, invalidate: true, tombstone_ttl: 30)
+
+        assert_predicate dc.get_with_status('tk'), :stale?
+      end
+    end
+  end
+
+  describe 'delete_multi with invalidate' do
+    it 'tombstones every key in the batch' do
+      memcached_persistent do |dc|
+        dc.flush
+        keys = %w[tm-a tm-b tm-c]
+        keys.each { |k| dc.set(k, "val-#{k}") }
+
+        deleted = dc.delete_multi(keys, invalidate: true, tombstone_ttl: 30)
+
+        assert_equal keys.length, deleted
+        keys.each do |k|
+          result = dc.get_with_status(k)
+
+          assert_predicate result, :stale?, "expected #{k} to be stale after delete_multi(invalidate: true)"
+          assert_equal "val-#{k}", result.value
+        end
+      end
+    end
+
+    it 'drops values across the batch when drop_value is set' do
+      memcached_persistent do |dc|
+        dc.flush
+        keys = %w[tm-x tm-y]
+        keys.each { |k| dc.set(k, 'orig') }
+
+        dc.delete_multi(keys, invalidate: true, tombstone_ttl: 30, drop_value: true)
+
+        keys.each do |k|
+          result = dc.get_with_status(k)
+
+          assert_predicate result, :stale?
+          assert_equal '', result.value
+        end
+      end
+    end
+  end
+
+  describe 'argument validation' do
+    it 'raises ArgumentError when tombstone_ttl is supplied without invalidate' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tk', 'val')
+
+        with_nil_logger do
+          assert_raises(ArgumentError) do
+            dc.delete('tk', tombstone_ttl: 30)
+          end
+        end
+      end
+    end
+  end
+
+  describe 'interaction with quiet block' do
+    it 'allows tombstone delete inside quiet (delete is in ALLOWED_QUIET_OPS)' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tq', 'val')
+
+        dc.quiet do
+          dc.delete('tq', invalidate: true, tombstone_ttl: 30)
+        end
+
+        # Outside the quiet block, the tombstone should be visible
+        assert_predicate dc.get_with_status('tq'), :stale?
+      end
+    end
+
+    it 'raises NotPermittedMultiOpError when get_with_status is called inside quiet' do
+      memcached_persistent do |dc|
+        dc.flush
+        dc.set('tq', 'val')
+
+        assert_raises(Dalli::NotPermittedMultiOpError) do
+          dc.quiet do
+            dc.get_with_status('tq')
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/test/protocol/meta/test_request_formatter.rb b/test/protocol/meta/test_request_formatter.rb
index a84ec4f5..ae23e7a0 100644
--- a/test/protocol/meta/test_request_formatter.rb
+++ b/test/protocol/meta/test_request_formatter.rb
@@ -198,6 +198,73 @@
                      key: key, p_token: 'route=a', l_token: 'hint=b'
                    )
     end
+
+    describe 'tombstone flags' do
+      it 'emits the I flag when invalidate is true' do
+        assert_equal "md #{key} I\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(key: key, invalidate: true)
+      end
+
+      it 'emits I T<n> when invalidate and tombstone_ttl are set' do
+        assert_equal "md #{key} I T30\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, invalidate: true, tombstone_ttl: 30
+                     )
+      end
+
+      it 'emits I T<n> x when invalidate, tombstone_ttl, and drop_value are all set' do
+        assert_equal "md #{key} I T30 x\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, invalidate: true, tombstone_ttl: 30, drop_value: true
+                     )
+      end
+
+      it 'emits I x without tombstone_ttl when invalidate and drop_value are set' do
+        assert_equal "md #{key} I x\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, invalidate: true, drop_value: true
+                     )
+      end
+
+      it 'allows drop_value alone (no invalidate) — only T-without-I is restricted' do
+        assert_equal "md #{key} x\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(key: key, drop_value: true)
+      end
+
+      it 'raises ArgumentError when tombstone_ttl is supplied without invalidate' do
+        assert_raises(ArgumentError) do
+          Dalli::Protocol::Meta::RequestFormatter.meta_delete(key: key, tombstone_ttl: 30)
+        end
+      end
+
+      it 'coerces a string tombstone_ttl into an integer (rejects non-numeric)' do
+        assert_equal "md #{key} I T30\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, invalidate: true, tombstone_ttl: '30'
+                     )
+
+        assert_raises(ArgumentError) do
+          Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+            key: key, invalidate: true, tombstone_ttl: "\nset importantkey 1 1000 8\ninjected"
+          )
+        end
+      end
+
+      it 'orders tombstone flags after q (quiet) and before routing tokens' do
+        assert_equal "md #{key} q I T30 x Proute=a Lhint=b\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, quiet: true, invalidate: true, tombstone_ttl: 30,
+                       drop_value: true, p_token: 'route=a', l_token: 'hint=b'
+                     )
+      end
+
+      it 'composes with cas (C<n> stays before tombstone tokens)' do
+        assert_equal "md #{key} C#{cas} I T30 x\r\n",
+                     Dalli::Protocol::Meta::RequestFormatter.meta_delete(
+                       key: key, cas: cas, invalidate: true, tombstone_ttl: 30, drop_value: true
+                     )
+      end
+    end
   end
 
   describe 'meta_arithmetic' do
diff --git a/test/test_opentelemetry_middleware.rb b/test/test_opentelemetry_middleware.rb
index a6face5f..a043166f 100644
--- a/test/test_opentelemetry_middleware.rb
+++ b/test/test_opentelemetry_middleware.rb
@@ -121,4 +121,95 @@
       assert_equal 1, attributes['miss_count'], 'expected miss_count to be 1 (one miss)'
     end
   end
+
+  it 'uses raw response bytes for get_with_status value_bytesize on non-string values' do
+    OTEL_EXPORTER.reset if OTEL_EXPORTER.respond_to?(:reset)
+
+    memcached(21_453, '', { middlewares: [Dalli::OpentelemetryMiddleware] }) do |dc, _|
+      test_value = { 'payload' => %w[a b c] }
+
+      assert op_addset_succeeds(dc.set('otel:status_hash', test_value, 30))
+
+      result = dc.get_with_status('otel:status_hash')
+
+      assert_equal test_value, result.value
+      assert_predicate result, :hit?
+      refute_predicate result, :stale?
+      refute_predicate result, :miss?
+
+      finished = OTEL_EXPORTER.respond_to?(:finished_spans) ? OTEL_EXPORTER.finished_spans : []
+      get_with_status_span = finished.find { |span| span.name == 'memcached.get_with_status' }
+
+      refute_nil get_with_status_span, 'expected to find a memcached.get_with_status span'
+
+      attributes = get_with_status_span.attributes
+
+      assert_equal Marshal.dump(test_value).bytesize, attributes['value_bytesize']
+      assert_equal 1, attributes['hit_count'], 'expected fresh result to count as a hit'
+      assert_equal 0, attributes['miss_count'], 'expected fresh result not to count as a miss'
+      assert_equal 0, attributes['stale_count'], 'expected fresh result not to count as stale'
+    end
+  end
+
+  it 'counts stale get_with_status results separately from fresh hits' do
+    OTEL_EXPORTER.reset if OTEL_EXPORTER.respond_to?(:reset)
+
+    memcached(21_453, '', { middlewares: [Dalli::OpentelemetryMiddleware] }) do |dc, _|
+      assert op_addset_succeeds(dc.set('otel:status_stale', 'stale-value', 30))
+      assert dc.delete('otel:status_stale', invalidate: true, tombstone_ttl: 30)
+
+      result = dc.get_with_status('otel:status_stale')
+
+      assert_predicate result, :hit?
+      assert_predicate result, :stale?
+      refute_predicate result, :miss?
+
+      finished = OTEL_EXPORTER.respond_to?(:finished_spans) ? OTEL_EXPORTER.finished_spans : []
+      get_with_status_span = finished.find { |span| span.name == 'memcached.get_with_status' }
+
+      refute_nil get_with_status_span, 'expected to find a memcached.get_with_status span'
+
+      attributes = get_with_status_span.attributes
+
+      assert_equal 0, attributes['hit_count'], 'expected stale result not to count as a fresh hit'
+      assert_equal 1, attributes['miss_count'], 'expected stale result to count as non-fresh'
+      assert_equal 1, attributes['stale_count'], 'expected stale_count to report the stale result'
+    end
+  end
+
+  it 'counts stale read_multi_with_status results separately from fresh hits' do
+    OTEL_EXPORTER.reset if OTEL_EXPORTER.respond_to?(:reset)
+
+    memcached(21_453, '', { middlewares: [Dalli::OpentelemetryMiddleware] }) do |dc, _|
+      assert op_addset_succeeds(dc.set('otel:status_hit', 'hit-value', 30))
+      assert op_addset_succeeds(dc.set('otel:status_stale', 'stale-value', 30))
+      assert op_addset_succeeds(dc.set('otel:status_dropped', 'dropped-value', 30))
+      assert dc.delete('otel:status_stale', invalidate: true, tombstone_ttl: 30)
+      assert dc.delete('otel:status_dropped', invalidate: true, drop_value: true, tombstone_ttl: 30)
+
+      results = dc.get_multi_with_status(
+        'otel:status_hit',
+        'otel:status_stale',
+        'otel:status_dropped',
+        'otel:status_absent'
+      )
+
+      assert_predicate results['otel:status_hit'], :hit?
+      refute_predicate results['otel:status_hit'], :stale?
+      assert_predicate results['otel:status_stale'], :stale?
+      assert_predicate results['otel:status_dropped'], :stale?
+      assert_predicate results['otel:status_absent'], :miss?
+
+      finished = OTEL_EXPORTER.respond_to?(:finished_spans) ? OTEL_EXPORTER.finished_spans : []
+      read_multi_with_status_span = finished.find { |span| span.name == 'memcached.read_multi_with_status' }
+
+      refute_nil read_multi_with_status_span, 'expected to find a memcached.read_multi_with_status span'
+
+      attributes = read_multi_with_status_span.attributes
+
+      assert_equal 1, attributes['hit_count'], 'expected only the fresh result to count as a hit'
+      assert_equal 3, attributes['miss_count'], 'expected stale and absent results to count as non-fresh'
+      assert_equal 2, attributes['stale_count'], 'expected stale_count to report both tombstones'
+    end
+  end
 end