diff --git a/lib/ldclient-rb/impl/file_data/reloader.rb b/lib/ldclient-rb/impl/file_data/reloader.rb
index a1166aa0..654eeb0e 100644
--- a/lib/ldclient-rb/impl/file_data/reloader.rb
+++ b/lib/ldclient-rb/impl/file_data/reloader.rb
@@ -83,8 +83,8 @@ class Reloader
         # @param on_error [#call, nil] invoked with the error when a reload fails, once per distinct
         #   failure. With automatic retries, repeats of an identical failure do not re-invoke it. A
         #   success re-arms it. The error is a {ReadError} when a file could not be read or parsed,
-        #   a {MergeError} when the documents could not be combined, or the exception that `apply`
-        #   raised. The reloader logs failures itself.
+        #   a {MergeError} when the documents could not be combined or `check_document` rejected
+        #   one, or the exception that `apply` raised. The reloader logs failures itself.
         # @param duplicate_keys_handling [Symbol] one of the {DuplicateKeysHandling} values
         # @param skip_missing_paths [Boolean] when true, a configured file that does not exist
         #   contributes no data, and the reload succeeds with the data of the files that exist.
@@ -98,6 +98,10 @@ class Reloader
         # @param skip_unchanged [Boolean] if true, `apply` is not invoked when the files' raw
         #   contents are byte-identical to the last successfully applied contents.
         # @param log_prefix [String] the prefix of every log line this reloader writes
+        # @param check_document [#call, nil] invoked with the path and the {Document} of each file
+        #   that a load parsed, before the documents are combined. A {MergeError} that it raises
+        #   fails the load in the same way as a file that cannot be parsed. When nil, every document
+        #   is combined.
         # @param clock [#call, nil] returns the seconds of a monotonic clock, for the time a file has
         #   been held back; nil for the system's. Scheduling always uses the system's clock.
         #
@@ -108,12 +112,14 @@ def initialize(paths:, logger:, apply:, on_error: nil,
                        retry_delay: DEFAULT_RETRY_DELAY,
                        skip_unchanged: false,
                        log_prefix: "[LDClient]",
+                       check_document: nil,
                        clock: nil)
           @paths = paths
           @logger = logger
           @log_prefix = log_prefix
           @apply = apply
           @on_error = on_error
+          @check_document = check_document
           @duplicate_keys_handling = duplicate_keys_handling
           @skip_missing_paths = skip_missing_paths
           @debounce_delay = debounce_delay
@@ -403,6 +409,12 @@ def stop
                 next
               end
             end
+            begin
+              @check_document&.call(path, document)
+            rescue MergeError => e
+              failure ||= e
+              next
+            end
             documents << document
             files << FileSummary.new(path, true, 0, 0)
           end
diff --git a/lib/ldclient-rb/impl/integrations/file_override_source.rb b/lib/ldclient-rb/impl/integrations/file_override_source.rb
index 6320d619..e9214f3d 100644
--- a/lib/ldclient-rb/impl/integrations/file_override_source.rb
+++ b/lib/ldclient-rb/impl/integrations/file_override_source.rb
@@ -71,6 +71,7 @@ def start(sink)
               duplicate_keys_handling: @duplicate_keys_handling,
               skip_missing_paths: true,
               skip_unchanged: true,
+              check_document: method(:reject_deleted_entries),
               apply: lambda do |merged|
                 sink.set_overrides(merged.flags.values, merged.segments.values)
                 log_overrides_in_effect(merged)
@@ -108,6 +109,28 @@ def stop
           reloader&.stop
         end
 
+        #
+        # Fails the load of a file that contains a flag or segment entry marked as deleted, in the
+        # same way as a file that cannot be parsed. An override is removed by removing its entry or
+        # deleting its file, so a deleted entry is a mistake in the file. The sink rejects such a
+        # snapshot too. Checking here lets the logged failure name the file and the key.
+        #
+        # @param path [String]
+        # @param document [LaunchDarkly::Impl::FileData::Document]
+        # @raise [LaunchDarkly::Impl::FileData::MergeError] if an entry is marked as deleted
+        #
+        private def reject_deleted_entries(path, document)
+          { "flag" => document.flags, "segment" => document.segments }.each do |name, entries|
+            entries.each do |key, data|
+              # The data model treats any value of "deleted" other than false or null as deleted.
+              next unless data.is_a?(Hash) && data[:deleted]
+
+              raise FileData::MergeError,
+                "#{name} \"#{key}\" in #{path} is marked as deleted; remove the entry to remove its override"
+            end
+          end
+        end
+
         #
         # Reports the overrides now in effect and the file each came from. The reloader applies a
         # snapshot only when the content changed, so this logs each change once.
diff --git a/lib/ldclient-rb/impl/overrides/sink.rb b/lib/ldclient-rb/impl/overrides/sink.rb
index f1e87064..be873dce 100644
--- a/lib/ldclient-rb/impl/overrides/sink.rb
+++ b/lib/ldclient-rb/impl/overrides/sink.rb
@@ -22,7 +22,9 @@ class Sink
         include LaunchDarkly::Interfaces::Overrides::OverrideSink
 
         KINDS = [DataStore::FEATURES, DataStore::SEGMENTS].freeze
-        private_constant :KINDS
+        # The name of each kind in messages, as users know it.
+        KIND_NAMES = { DataStore::FEATURES => "flag", DataStore::SEGMENTS => "segment" }.freeze
+        private_constant :KINDS, :KIND_NAMES
 
         #
         # @param layer [Layer] the layer to write to
@@ -184,23 +186,35 @@ def self.affected_flag_keys(previous, current, old_merged, new_merged)
 
         #
         # Converts the definitions passed to the sink into a hash keyed by symbol. A hash entry is
-        # deserialized into the data model. A model object is used as is. A deleted entry is
-        # rejected before the layer is replaced: an override defines a flag or segment, and a
-        # deleted one would shadow the LaunchDarkly item with one that cannot be evaluated.
+        # deserialized into the data model. A model object is used as is. A snapshot with a deleted
+        # entry is rejected before the layer is replaced: an override defines a flag or segment, and
+        # a deleted one would shadow the LaunchDarkly item with one that cannot be evaluated.
         #
         private def index_items(kind, items)
+          name = KIND_NAMES[kind]
           result = {}
           (items || []).each do |item|
             model = Model.deserialize(kind, item, @logger)
             key = model.respond_to?(:key) ? model.key : nil
-            raise ArgumentError, "an override #{kind.namespace} entry has no key" if key.nil? || key.to_s.empty?
-            raise ArgumentError, "the override #{kind.namespace} entry #{key} is deleted" if model.respond_to?(:deleted) && model.deleted
+            raise ArgumentError, "an override #{name} has no key" if key.nil? || key.to_s.empty?
+            reject_deleted(name, key) if model.respond_to?(:deleted) && model.deleted
 
             result[key.to_sym] = model
           end
           result
         end
 
+        #
+        # Rejects a snapshot that contains a deleted item. The rejection is logged as well as
+        # raised: a source that supplies a snapshot after it has started may not log the error it
+        # receives.
+        #
+        private def reject_deleted(name, key)
+          message = "override #{name} \"#{key}\" is marked as deleted; leave the #{name} out of the snapshot to remove its override"
+          @logger.error { "[LDClient] Rejected an override snapshot, keeping the overrides in effect: #{message}" }
+          raise ArgumentError, message
+        end
+
         #
         # Captures the merged view of the base store and a layer snapshot: base data with override
         # entries overlaid. A base read failure for a kind yields just the overrides for that kind.
diff --git a/lib/ldclient-rb/integrations/file_data.rb b/lib/ldclient-rb/integrations/file_data.rb
index 8b3dbaf3..ebd82648 100644
--- a/lib/ldclient-rb/integrations/file_data.rb
+++ b/lib/ldclient-rb/integrations/file_data.rb
@@ -186,8 +186,10 @@ def self.data_source_v2(options = {})
       # later, and deleting a file removes its overrides. A file that exists but cannot be read or
       # parsed fails that whole reload. The previously loaded overrides stay in effect, the failure is
       # logged, and the source retries after a short delay and recovers on its own once the file is
-      # readable. Whenever the set of overrides in effect changes, including at startup, the source
-      # logs the overrides in effect and what each file supplied, at Info level.
+      # readable. A file that contains an entry marked as deleted fails to load, and the overrides in
+      # effect stay; to remove an override, remove its entry or delete the file. Whenever the set of
+      # overrides in effect changes, including at startup, the source logs the overrides in effect
+      # and what each file supplied, at Info level.
       #
       # Files can be rewritten in place: if the source finds a file empty while it is being rewritten,
       # it keeps the overrides in effect until the new content is written. To remove a file's
diff --git a/lib/ldclient-rb/interfaces/overrides.rb b/lib/ldclient-rb/interfaces/overrides.rb
index 431d9658..d98e4c29 100644
--- a/lib/ldclient-rb/interfaces/overrides.rb
+++ b/lib/ldclient-rb/interfaces/overrides.rb
@@ -25,7 +25,8 @@ module OverrideSink
         #
         # Each definition is either a data model object, as produced by
         # `LaunchDarkly::Impl::Model.deserialize`, or a hash in the flag or segment data model.
-        # The SDK itself marks the entries as overrides.
+        # The SDK itself marks the entries as overrides. A snapshot that contains a flag or segment
+        # marked as deleted is rejected with `ArgumentError`, and the store keeps its contents.
         #
         # This method is safe to call from any thread. Calls are serialized by the SDK, and the
         # new contents are visible to evaluations when the call returns.
diff --git a/spec/impl/file_data/reloader_spec.rb b/spec/impl/file_data/reloader_spec.rb
index 404f71b2..8c810cb1 100644
--- a/spec/impl/file_data/reloader_spec.rb
+++ b/spec/impl/file_data/reloader_spec.rb
@@ -206,6 +206,41 @@ def with_reloader(paths, logger: $null_log, reject: 0, **options)
           end
         end
 
+        it "keeps the last good result when the document check rejects a file, and still checks the other files" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          b = write("b.json", values_doc({ flag2: "b" }))
+          logger = CapturingLogger.new
+          checked = []
+          check = lambda do |path, document|
+            checked << path
+            raise MergeError, "#{path} has a bad entry" if document.flag_values.key?(:bad)
+          end
+
+          with_reloader([a, b], logger: logger, retry_delay: 0, check_document: check) do |reloader, recorder|
+            expect(reloader.reload_now).to be true
+            write("a.json", values_doc({ flag1: "changed", bad: true }))
+
+            expect(reloader.reload_now).to be false
+            expect(recorder.applied.length).to eq(1)
+            expect(recorder.flag_values).to eq({ flag1: "a", flag2: "b" })
+            expect(recorder.errors.map(&:message)).to eq(["#{a} has a bad entry"])
+            expect(checked).to eq([a, b, a, b])
+            expect(logger.output).to include("ERROR -- : [LDClient] Unable to load flags: #{a} has a bad entry")
+          end
+        end
+
+        it "combines entries marked as deleted when no document check is given" do
+          # The override source rejects such entries through its document check. Without one, the
+          # reloader combines them as before.
+          a = write("a.json", { flags: { flag1: { on: false, deleted: true } }, segments: { seg1: { deleted: true } } }.to_json)
+
+          with_reloader([a]) do |reloader, recorder|
+            expect(reloader.reload_now).to be true
+            expect(recorder.applied[0].flags[:flag1].deleted).to be true
+            expect(recorder.applied[0].segments[:seg1].deleted).to be true
+          end
+        end
+
         it "retries a failed reload after the retry delay and recovers when the file is fixed" do
           a = write("a.json", '{"flagValues"')
 
diff --git a/spec/impl/overrides/sink_spec.rb b/spec/impl/overrides/sink_spec.rb
index 21b991bc..f628155a 100644
--- a/spec/impl/overrides/sink_spec.rb
+++ b/spec/impl/overrides/sink_spec.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require "spec_helper"
+require "capturing_logger"
 require "model_builders"
 require "ldclient-rb/impl/overrides"
 require "ldclient-rb/impl/broadcaster"
@@ -15,7 +16,8 @@ module Overrides
         let(:broadcaster) { Broadcaster.new(SynchronousExecutor.new, $null_log) }
         let(:base) { DataStore::InMemoryFeatureStoreV2.new($null_log) }
         let(:layer) { Layer.new }
-        subject(:sink) { Sink.new(layer, base, broadcaster, $null_log) }
+        let(:logger) { CapturingLogger.new }
+        subject(:sink) { Sink.new(layer, base, broadcaster, logger) }
 
         def flag(key, value, version: 1, prereqs: [], segment: nil)
           data = {
@@ -77,15 +79,45 @@ def changed_keys(listener)
           expect { sink.set_overrides([], [{ key: "", version: 1 }]) }.to raise_error(ArgumentError, /has no key/)
         end
 
-        it "rejects a deleted entry before the layer is replaced" do
-          sink.set_overrides([flag("a", "x")], [])
+        it "rejects the whole snapshot when a flag is deleted, keeps the layer, and logs the flag" do
+          sink.set_overrides([flag("a", "x")], [segment("s")])
           listener = listen
 
-          expect { sink.set_overrides([{ key: "f", version: 2, deleted: true }], []) }.to raise_error(ArgumentError, /is deleted/)
-          expect { sink.set_overrides([], [{ key: "s", version: 2, deleted: true }]) }.to raise_error(ArgumentError, /is deleted/)
+          # The snapshot's other entries are not applied either.
+          expect { sink.set_overrides([flag("a", "y"), { key: "f", version: 2, deleted: true }], []) }
+            .to raise_error(ArgumentError, 'override flag "f" is marked as deleted; leave the flag out of the snapshot to remove its override')
           expect(layer.get(DataStore::FEATURES, "a").variations).to eq ["x"]
+          expect(layer.get(DataStore::SEGMENTS, "s")).not_to be_nil
           expect(layer.get(DataStore::FEATURES, "f")).to be_nil
           expect(listener.statuses).to be_empty
+          expect(logger.output).to include(
+            'ERROR -- : [LDClient] Rejected an override snapshot, keeping the overrides in effect: override flag "f" is marked as deleted'
+          )
+        end
+
+        it "rejects the whole snapshot when a segment is deleted, keeps the layer, and logs the segment" do
+          sink.set_overrides([flag("a", "x")], [segment("s")])
+          listener = listen
+
+          expect { sink.set_overrides([flag("b", "y")], [segment("t"), Segments.from_hash({ key: "gone", version: 2, deleted: true })]) }
+            .to raise_error(ArgumentError, /\Aoverride segment "gone" is marked as deleted/)
+          expect(layer.get(DataStore::FEATURES, "a").variations).to eq ["x"]
+          expect(layer.get(DataStore::FEATURES, "b")).to be_nil
+          expect(layer.get(DataStore::SEGMENTS, "s")).not_to be_nil
+          expect(layer.get(DataStore::SEGMENTS, "t")).to be_nil
+          expect(listener.statuses).to be_empty
+          expect(logger.output).to include(
+            'ERROR -- : [LDClient] Rejected an override snapshot, keeping the overrides in effect: override segment "gone" is marked as deleted'
+          )
+        end
+
+        it "accepts entries whose deleted property is false" do
+          sink.set_overrides([{ key: "f", version: 1, deleted: false, on: false, offVariation: 0, variations: ["x"] }],
+            [{ key: "s", version: 1, deleted: false, included: ["user"] }])
+
+          expect(layer.get(DataStore::FEATURES, "f").variations).to eq ["x"]
+          expect(layer.get(DataStore::SEGMENTS, "s")).not_to be_nil
+          expect(logger.output).not_to include("ERROR")
         end
 
         it "notifies every changed entry when an entry's dependency fields cannot be read" do
diff --git a/spec/integrations/file_data_source_spec.rb b/spec/integrations/file_data_source_spec.rb
index 410d8e4e..4ba8e9b1 100644
--- a/spec/integrations/file_data_source_spec.rb
+++ b/spec/integrations/file_data_source_spec.rb
@@ -271,6 +271,23 @@ def with_data_source(options, initialize_to_valid = false)
         end
       end
 
+      it "loads entries marked as deleted into the data store" do
+        # Unlike the file override source, the data source does not reject such entries. With no
+        # update sink, it writes to the data store directly, which holds them as deleted items.
+        file = make_temp_file('{"flags": {"flag1": {"key": "flag1", "deleted": true}, ' \
+          '"flag2": {"key": "flag2", "on": false, "variations": ["x"]}}}')
+        config = LaunchDarkly::Config.new(logger: $null_log)
+        ds = FileData.data_source({ paths: [ file.path ] }).call('', config)
+        begin
+          ds.start
+          expect(ds.initialized?).to eq(true)
+          expect(config.feature_store.get(LaunchDarkly::Impl::DataStore::FEATURES, "flag1")).to be_nil
+          expect(config.feature_store.get(LaunchDarkly::Impl::DataStore::FEATURES, "flag2")).not_to be_nil
+        ensure
+          ds.stop
+        end
+      end
+
       it "does not reload modified file if auto-update is off" do
         file = make_temp_file(flag_only_json)
 
diff --git a/spec/integrations/file_data_source_v2_spec.rb b/spec/integrations/file_data_source_v2_spec.rb
index 670e9adf..2f157999 100644
--- a/spec/integrations/file_data_source_v2_spec.rb
+++ b/spec/integrations/file_data_source_v2_spec.rb
@@ -256,6 +256,26 @@ def no_selector_store
             source.stop
           end
         end
+
+        it "loads entries marked as deleted" do
+          # Unlike the file override source, the data source does not reject such entries.
+          file = make_temp_file('{"flags": {"flag1": {"key": "flag1", "version": 2, "deleted": true}},' \
+            ' "segments": {"seg1": {"key": "seg1", "deleted": true}}}')
+
+          source = Impl::Integrations::FileDataSourceV2.new(logger, paths: [file.path])
+
+          begin
+            result = source.fetch(no_selector_store)
+
+            expect(result.success?).to eq(true)
+            changes = result.value.change_set.changes
+            kinds = LaunchDarkly::Interfaces::DataSystem::ObjectKind
+            expect(changes.map { |c| [c.kind, c.key, c.object[:deleted]] })
+              .to eq([[kinds::FLAG, :flag1, true], [kinds::SEGMENT, :seg1, true]])
+          ensure
+            source.stop
+          end
+        end
       end
 
       describe "synchronizer (sync)" do
diff --git a/spec/integrations/file_override_source_spec.rb b/spec/integrations/file_override_source_spec.rb
index 4df47d21..bbc7a894 100644
--- a/spec/integrations/file_override_source_spec.rb
+++ b/spec/integrations/file_override_source_spec.rb
@@ -287,6 +287,70 @@ def with_source(paths, **options)
         end
       end
 
+      describe "an entry marked as deleted" do
+        def deleted_message(category, key, file)
+          %(#{category} "#{key}" in #{file} is marked as deleted; remove the entry to remove its override)
+        end
+
+        it "fails the initial load, leaving the sink untouched, and logs the file and the flag" do
+          a = write("a.json", { flagValues: { flag1: "a" }, flags: { flag2: { on: false, variations: ["x"], deleted: true } } }.to_json)
+
+          with_source([a]) do |_source, sink|
+            expect(sink.snapshots).to be_empty
+            expect(logger.output).to include(
+              "ERROR -- : [LDClient] FileOverrideSource: Unable to load flags: #{deleted_message('flag', 'flag2', a)}"
+            )
+          end
+        end
+
+        it "keeps the overrides in effect when a flag is marked as deleted, and applies the fix" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            write("a.json", { flagValues: { flag1: "b" }, flags: { flag2: { deleted: true } } }.to_json)
+            expect(wait_for { logger.output.include?(deleted_message("flag", "flag2", a)) }).to be true
+            expect(sink.snapshots.length).to eq 1
+            expect(sink.flag_values).to eq({ "flag1" => "a" })
+
+            write("a.json", values_doc({ flag1: "c" }))
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "c" })
+          end
+        end
+
+        it "keeps every file's overrides in effect while a segment in one file is marked as deleted" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          b = write("b.yaml", "segments:\n  seg1:\n    included: [user]\n")
+
+          with_source([a, b]) do |_source, sink|
+            write("b.yaml", "segments:\n  seg1:\n    deleted: true\n")
+            expect(wait_for { logger.output.include?(deleted_message("segment", "seg1", b)) }).to be true
+
+            # A change to the other file is not applied either while the reload fails.
+            write("a.json", values_doc({ flag1: "b" }))
+            sleep 0.5
+            expect(sink.snapshots.length).to eq 1
+            expect(sink.segment_keys).to eq ["seg1"]
+
+            write("b.yaml", "segments: {}\n")
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "b" })
+            expect(sink.segment_keys).to eq []
+          end
+        end
+
+        it "loads entries whose deleted property is false" do
+          a = write("a.json", { flags: { flag1: { on: false, offVariation: 0, variations: ["x"], deleted: false } },
+                                segments: { seg1: { included: ["user"], deleted: false } } }.to_json)
+
+          with_source([a]) do |_source, sink|
+            expect(sink.flag_values).to eq({ "flag1" => "x" })
+            expect(sink.segment_keys).to eq ["seg1"]
+            expect(logger.output).not_to include("ERROR")
+          end
+        end
+      end
+
       describe "logging" do
         it "logs the overrides in effect and what each file supplied" do
           a = write("a.json", { flagValues: { flag1: "a", flag2: "b" }, segments: { seg1: { key: "seg1" } } }.to_json)
@@ -693,6 +757,23 @@ def with_client(overrides)
           end
         end
 
+        it "keeps serving an override when its file marks the flag as deleted, and names the file in the log" do
+          a = write("a.json", values_doc({ flag: "override-value" }))
+          source = make_source([a])
+          builder = Object.new
+          builder.define_singleton_method(:build) { |_sdk_key, _config| source }
+
+          with_client(builder) do |client|
+            expect(client.variation("flag", context, "default")).to eq "override-value"
+            write("a.json", { flags: { flag: { deleted: true } } }.to_json)
+
+            expect(wait_for { logger.output.include?(%(flag "flag" in #{a} is marked as deleted)) }).to be true
+            expect(client.variation("flag", context, "default")).to eq "override-value"
+            # The source fails the load before it supplies the snapshot, so the client does not reject it.
+            expect(logger.output).not_to include("Rejected an override snapshot")
+          end
+        end
+
         it "reports invalid options from client construction" do
           data_system = DataSystem.custom.overrides(FileData.override_source(paths: [])).build