diff --git a/docs/metasploit-framework.wiki/How-to-use-datastore-options.md b/docs/metasploit-framework.wiki/How-to-use-datastore-options.md
index e76a70d52c75..618b301ae84b 100644
--- a/docs/metasploit-framework.wiki/How-to-use-datastore-options.md
+++ b/docs/metasploit-framework.wiki/How-to-use-datastore-options.md
@@ -86,8 +86,7 @@ OptSomething.new(option_name, [boolean, description, value, *enums*], aliases: *
   options](#Filtering-datastore-options) section for more information.
 * **fallbacks** *optional*, *key-word only* An array of names that will be used as a fallback if the main option name is
   defined by the user. This is useful in the scenario of wanting specialised option names such as `SMBUser`, but to also
-  support gracefully checking a list of more generic fallbacks option names such as `Username`. This functionality is 
-  currently behind a feature flag, set with `features set datastore_fallbacks true` in msfconsole
+  support gracefully checking a list of more generic fallbacks option names such as `Username`.
 
 Now let's talk about what classes are available:
 
diff --git a/lib/msf/base/serializer/readable_text.rb b/lib/msf/base/serializer/readable_text.rb
index c3710f7156db..9d49ecdd6079 100644
--- a/lib/msf/base/serializer/readable_text.rb
+++ b/lib/msf/base/serializer/readable_text.rb
@@ -623,7 +623,7 @@ def self.options_table(missing, mod, options, indent)
     )
     options.sort_by(&:name).each do |opt|
       name = opt.name
-      if mod.datastore.is_a?(Msf::DataStoreWithFallbacks)
+      if mod.datastore.is_a?(Msf::DataStore)
         val = mod.datastore[name]
       else
         val = mod.datastore[name].nil? ? opt.default : mod.datastore[name]
diff --git a/lib/msf/core/data_store.rb b/lib/msf/core/data_store.rb
index ca9c6c022da5..d7bac00ac35b 100644
--- a/lib/msf/core/data_store.rb
+++ b/lib/msf/core/data_store.rb
@@ -3,40 +3,61 @@ module Msf
 
 ###
 #
-# The data store is just a bitbucket that holds keyed values.  It is used
+# The data store is just a bitbucket that holds keyed values. It is used
 # by various classes to hold option values and other state information.
 #
 ###
-class DataStore < Hash
-
-  # Temporary forking logic for conditionally using the {Msf::ModuleDatastoreWithFallbacks} implementation.
-  #
-  # This method replaces the default `ModuleDataStore.new` with the ability to instantiate the `ModuleDataStoreWithFallbacks`
-  # class instead, if the feature is enabled
-  def self.new
-    if Msf::FeatureManager.instance.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-      return Msf::DataStoreWithFallbacks.new
-    end
-
-    instance = allocate
-    instance.send(:initialize)
-    instance
-  end
+class DataStore
+
+  # The global framework datastore doesn't currently import options
+  # For now, store an ad-hoc list of keys that the shell handles
+  #
+  # This list could be removed if framework's bootup sequence registers
+  # these as datastore options
+  GLOBAL_KEYS = %w[
+    ConsoleLogging
+    LogLevel
+    MinimumRank
+    SessionLogging
+    TimestampOutput
+    Prompt
+    PromptChar
+    PromptTimeFormat
+    MeterpreterPrompt
+    SessionTlvLogging
+  ]
 
   #
   # Initializes the data store's internal state.
   #
-  def initialize()
+  def initialize
     @options     = Hash.new
     @aliases     = Hash.new
-    @imported    = Hash.new
-    @imported_by = Hash.new
+
+    # default values which will be referenced when not defined by the user
+    @defaults = Hash.new
+
+    # values explicitly defined, which take precedence over default values
+    @user_defined = Hash.new
   end
 
+  # @return [Hash{String => Msf::OptBase}] The options associated with this datastore. Used for validating values/defaults/etc
   attr_accessor :options
-  attr_accessor :aliases
-  attr_accessor :imported
-  attr_accessor :imported_by
+
+  #
+  # Returns a hash of user-defined datastore values. The returned hash does
+  # not include default option values.
+  #
+  # @return [Hash<String, Object>] values explicitly defined on the data store which will override any default datastore values
+  attr_accessor :user_defined
+
+  #
+  # Was this entry actually set or just using its default
+  #
+  # @return [TrueClass, FalseClass]
+  def default?(key)
+    search_for(key).default?
+  end
 
   #
   # Clears the imported flag for the supplied key since it's being set
@@ -44,8 +65,6 @@ def initialize()
   #
   def []=(k, v)
     k = find_key_case(k)
-    @imported[k] = false
-    @imported_by[k] = nil
 
     opt = @options[k]
     unless opt.nil?
@@ -57,49 +76,76 @@ def []=(k, v)
       end
     end
 
-    super(k,v)
+    @user_defined[k] = v
   end
 
   #
   # Case-insensitive wrapper around hash lookup
   #
   def [](k)
-    super(find_key_case(k))
+    search_result = search_for(k)
+
+    search_result.value
   end
 
   #
-  # Case-insensitive wrapper around store
+  # Case-insensitive wrapper around store; Skips option validation entirely
   #
   def store(k,v)
-    super(find_key_case(k), v)
+    @user_defined[find_key_case(k)] = v
   end
 
   #
-  # Case-insensitive wrapper around delete
+  # Updates a value in the datastore with the specified name, k, to the
+  # specified value, v. Skips option validation entirely.
+  #
+  def update_value(k, v)
+    store(k, v)
+  end
+
   #
-  def delete(k)
-    @aliases.delete_if { |_, v| v.casecmp(k) == 0 }
-    super(find_key_case(k))
+  # unset the current key from the datastore
+  # @param [String] key The key to search for
+  def unset(key)
+    k = find_key_case(key)
+    search_result = search_for(k)
+    @user_defined.delete(k)
+
+    search_result.value
   end
 
+  # @deprecated use #{unset} instead, or set the value explicitly to nil
+  # @param [String] key The key to search for
+  def delete(key)
+    unset(key)
+  end
 
   #
-  # Updates a value in the datastore with the specified name, k, to the
-  # specified value, v.  This update does not alter the imported status of
-  # the value.
+  # Removes an option and any associated value
   #
-  def update_value(k, v)
-    self.store(k, v)
+  # @param [String] name the option name
+  # @return [nil]
+  def remove_option(name)
+    k = find_key_case(name)
+    @user_defined.delete(k)
+    @aliases.delete_if { |_, v| v.casecmp?(k) }
+    @options.delete_if { |option_name, _v| option_name.casecmp?(k) || option_name.casecmp?(name) }
+
+    nil
   end
 
   #
   # This method is a helper method that imports the default value for
   # all of the supplied options
   #
-  def import_options(options, imported_by = nil, overwrite = false)
-    options.each_option do |name, opt|
-      if self[name].nil? || overwrite
-        import_option(name, opt.default, true, imported_by, opt)
+  def import_options(options, imported_by = nil, overwrite = true)
+    options.each_option do |name, option|
+      if self.options[name].nil? || overwrite
+        key = name
+        option.aliases.each do |a|
+          @aliases[a.downcase] = key.downcase
+        end
+        @options[key] = option
       end
     end
   end
@@ -142,22 +188,32 @@ def import_options_from_s(option_str, delim = nil)
       hash[var] = val
     }
 
-    import_options_from_hash(hash)
+    merge!(hash)
   end
 
   #
-  # Imports options from a hash and stores them in the datastore.
+  # Imports values from a hash and stores them in the datastore.
   #
+  # @deprecated use {#merge!} instead
+  # @return [nil]
   def import_options_from_hash(option_hash, imported = true, imported_by = nil)
-    option_hash.each_pair { |key, val|
-      import_option(key, val, imported, imported_by)
-    }
+    merge!(option_hash)
+  end
+
+  # Update defaults from a hash. These merged values are not validated by default.
+  #
+  # @param [Hash<String, Object>] hash The default values that should be used by the datastore
+  # @param [Object] imported_by Who imported the defaults, not currently used
+  # @return [nil]
+  def import_defaults_from_hash(hash, imported_by:)
+    @defaults.merge!(hash)
   end
 
   # TODO: Doesn't normalize data in the same vein as:
   # https://github.com/rapid7/metasploit-framework/pull/6644
+  # @deprecated Use {#import_options}
   def import_option(key, val, imported = true, imported_by = nil, option = nil)
-    self.store(key, val)
+    store(key, val)
 
     if option
       option.aliases.each do |a|
@@ -165,10 +221,32 @@ def import_option(key, val, imported = true, imported_by = nil, option = nil)
       end
     end
     @options[key] = option
-    @imported[key] = imported
-    @imported_by[key] = imported_by
   end
 
+  # @return [Array<String>] The array of user defined datastore values, and registered option names
+  def keys
+    (@user_defined.keys + @options.keys).uniq(&:downcase)
+  end
+
+  # @return [Integer] The length of the registered keys
+  def length
+    keys.length
+  end
+
+  alias count length
+  alias size length
+
+  # @param [String] key
+  # @return [TrueClass, FalseClass] True if the key is present in the user defined values, or within registered options. False otherwise.
+  def key?(key)
+    matching_key = find_key_case(key)
+    keys.include?(matching_key)
+  end
+
+  alias has_key? key?
+  alias include? key?
+  alias member? key?
+
   #
   # Serializes the options in the datastore to a string.
   #
@@ -179,7 +257,7 @@ def to_s(delim = ' ')
       str << "#{key}=#{self[key]}" + ((str.length) ? delim : '')
     }
 
-    return str
+    str
   end
 
   # Override Hash's to_h method so we can include the original case of each key
@@ -225,7 +303,7 @@ def to_file(path, name = 'global')
     ini.add_group(name)
 
     # Save all user-defined options to the file.
-    user_defined.each_pair { |k, v|
+    @user_defined.each_pair { |k, v|
       ini[name][k] = v
     }
 
@@ -243,73 +321,73 @@ def from_file(path, name = 'global')
       return
     end
 
-    if (ini.group?(name))
-      import_options_from_hash(ini[name], false)
+    if ini.group?(name)
+      merge!(ini[name])
     end
   end
 
   #
-  # Return a deep copy of this datastore.
-  #
+  # Return a copy of this datastore. Only string values will be duplicated, other values
+  # will share the same reference
+  # @return [Msf::DataStore] a new datastore instance
   def copy
-    ds = self.class.new
-    self.keys.each do |k|
-      ds.import_option(k, self[k].kind_of?(String) ? self[k].dup : self[k], @imported[k], @imported_by[k])
-    end
-    ds.aliases = self.aliases.dup
-    ds
+    new_instance = self.class.new
+    new_instance.copy_state(self)
+    new_instance
   end
 
   #
-  # Override merge! so that we merge the aliases and imported hashes
+  # Merge the other object into the current datastore's aliases and imported hashes
   #
+  # @param [Msf::Datastore, Hash] other
   def merge!(other)
-    if other.is_a? DataStore
+    if other.is_a?(DataStore)
       self.aliases.merge!(other.aliases)
-      self.imported.merge!(other.imported)
-      self.imported_by.merge!(other.imported_by)
+      self.options.merge!(other.options)
+      self.defaults.merge!(other.defaults)
+      other.user_defined.each do |k, v|
+        @user_defined[find_key_case(k)] = v
+      end
+    else
+      other.each do |k, v|
+        self.store(k, v)
+      end
     end
-    # call super last so that we return a reference to ourselves
-    super
-  end
 
-  #
-  # Override merge to ensure we merge the aliases and imported hashes
-  #
-  def merge(other)
-    ds = self.copy
-    ds.merge!(other)
+    self
   end
 
+  alias update merge!
+
   #
-  # Returns a hash of user-defined datastore values.  The returned hash does
-  # not include default option values.
+  # Reverse Merge the other object into the current datastore's aliases and imported hashes
+  # Equivalent to ActiveSupport's reverse_merge! functionality.
   #
-  def user_defined
-    reject { |k, v|
-      @imported[k] == true
-    }
+  # @param [Msf::Datastore] other
+  def reverse_merge!(other)
+    raise ArgumentError, "invalid error type #{other.class}, expected ::Msf::DataStore" unless other.is_a?(Msf::DataStore)
+
+    copy_state(other.merge(self))
   end
 
   #
-  # Remove all imported options from the data store.
+  # Override merge to ensure we merge the aliases and imported hashes
   #
-  def clear_non_user_defined
-    @imported.delete_if { |k, v|
-      if (v and @imported_by[k] != 'self')
-        self.delete(k)
-        @imported_by.delete(k)
-      end
-
-      v
-    }
+  # @param [Msf::Datastore,Hash] other
+  def merge(other)
+    ds = self.copy
+    ds.merge!(other)
   end
 
   #
-  # Completely clear all values in the hash
+  # Completely clear all values in the data store
   #
   def clear
-    self.keys.each {|k| self.delete(k) }
+    self.options.clear
+    self.aliases.clear
+    self.defaults.clear
+    self.user_defined.clear
+
     self
   end
 
@@ -325,28 +403,145 @@ def each(&block)
     list.each(&block)
   end
 
+  alias each_pair each
+
+  def each_key(&block)
+    self.keys.each(&block)
+  end
+
   #
   # Case-insensitive key lookup
   #
+  # @return [String]
   def find_key_case(k)
-
     # Scan each alias looking for a key
     search_k = k.downcase
     if self.aliases.has_key?(search_k)
       search_k = self.aliases[search_k]
     end
 
+    # Check to see if we have an exact key match - otherwise we'll have to search manually to check case sensitivity
+    if @user_defined.key?(search_k) || options.key?(search_k)
+      return search_k
+    end
+
     # Scan each key looking for a match
-    self.each_key do |rk|
+    each_key do |rk|
       if rk.casecmp(search_k) == 0
         return rk
       end
     end
 
     # Fall through to the non-existent value
-    return k
+    k
+  end
+
+  # Search for a value within the current datastore, taking into consideration any registered aliases, fallbacks, etc.
+  #
+  # @param [String] key The key to search for
+  # @return [DataStoreSearchResult]
+  def search_for(key)
+    k = find_key_case(key)
+    return search_result(:user_defined, @user_defined[k]) if @user_defined.key?(k)
+
+    option = @options.fetch(k) { @options.find { |option_name, _option| option_name.casecmp?(k) }&.last }
+    if option
+      # If the key isn't present - check any additional fallbacks that have been registered with the option.
+      # i.e. handling the scenario of SMBUser not being explicitly set, but the option has registered a more
+      # generic 'Username' fallback
+      option.fallbacks.each do |fallback|
+      fallback_search = search_for(fallback)
+        if fallback_search.found?
+          return search_result(:option_fallback, fallback_search.value, fallback_key: fallback)
+        end
+      end
+    end
+
+    # Checking for imported default values, ignoring case again
+    imported_default_match = @defaults.find { |default_key, _default_value| default_key.casecmp?(k) }
+    return search_result(:imported_default, imported_default_match.last) if imported_default_match
+    return search_result(:option_default, option.default) if option
+
+    search_result(:not_found, nil)
+  end
+
+  protected
+
+  # These defaults will be used if the user has not explicitly defined a specific datastore value.
+  # These will be checked as a priority to any options that also provide defaults.
+  #
+  # @return [Hash{String => Msf::OptBase}] The hash of default values
+  attr_accessor :defaults
+
+  # @return [Hash{String => String}] The key is the old option name, the value is the new option name
+  attr_accessor :aliases
+
+  #
+  # Copy the state from the other Msf::DataStore. The state will be coped in a shallow fashion, other than
+  # imported and user_defined strings.
+  #
+  # @param [Msf::DataStore] other The other datastore to copy state from
+  # @return [Msf::DataStore] the current datastore instance
+  def copy_state(other)
+    self.options = other.options.dup
+    self.aliases = other.aliases.dup
+    self.defaults = other.defaults.transform_values { |value| value.kind_of?(String) ? value.dup : value }
+    self.user_defined = other.user_defined.transform_values { |value| value.kind_of?(String) ? value.dup : value }
+
+    self
+  end
+
+  # Raised when the specified key is not found
+  # @param [string] key
+  def key_error_for(key)
+    ::KeyError.new "key not found: #{key.inspect}"
+  end
+
+  #
+  # Simple dataclass for storing the result of a datastore search
+  #
+  class DataStoreSearchResult
+    # @return [String, nil] the key associated with the fallback value
+    attr_reader :fallback_key
+
+    # @return [object, nil] The value if found
+    attr_reader :value
+
+    def initialize(result, value, namespace: nil, fallback_key: nil)
+      @namespace = namespace
+      @result = result
+      @value = value
+      @fallback_key = fallback_key
+    end
+
+    def default?
+      result == :imported_default || result == :option_default || !found?
+    end
+
+    def found?
+      result != :not_found
+    end
+
+    def fallback?
+      result == :option_fallback
+    end
+
+    def global?
+      namespace == :global_data_store && found?
+    end
+
+  protected
+
+    # @return [Symbol] namespace Where the search result was found, i.e. a module datastore or global datastore
+    attr_reader :namespace
+
+    # @return [Symbol] result is one of `user_defined`, `not_found`, `option_fallback`, `option_default`, `imported_default`
+    attr_reader :result
   end
 
+  def search_result(result, value, fallback_key: nil)
+    DataStoreSearchResult.new(result, value, namespace: :global_data_store, fallback_key: fallback_key)
+  end
 end
 
 end
diff --git a/lib/msf/core/data_store_with_fallbacks.rb b/lib/msf/core/data_store_with_fallbacks.rb
deleted file mode 100644
index 7d6d372a72de..000000000000
--- a/lib/msf/core/data_store_with_fallbacks.rb
+++ /dev/null
@@ -1,547 +0,0 @@
-# -*- coding: binary -*-
-module Msf
-
-###
-#
-# The data store is just a bitbucket that holds keyed values. It is used
-# by various classes to hold option values and other state information.
-#
-###
-class DataStoreWithFallbacks
-
-  # The global framework datastore doesn't currently import options
-  # For now, store an ad-hoc list of keys that the shell handles
-  #
-  # This list could be removed if framework's bootup sequence registers
-  # these as datastore options
-  GLOBAL_KEYS = %w[
-    ConsoleLogging
-    LogLevel
-    MinimumRank
-    SessionLogging
-    TimestampOutput
-    Prompt
-    PromptChar
-    PromptTimeFormat
-    MeterpreterPrompt
-    SessionTlvLogging
-  ]
-
-  #
-  # Initializes the data store's internal state.
-  #
-  def initialize
-    @options     = Hash.new
-    @aliases     = Hash.new
-
-    # default values which will be referenced when not defined by the user
-    @defaults = Hash.new
-
-    # values explicitly defined, which take precedence over default values
-    @user_defined = Hash.new
-  end
-
-  # @return [Hash{String => Msf::OptBase}] The options associated with this datastore. Used for validating values/defaults/etc
-  attr_accessor :options
-
-  #
-  # Returns a hash of user-defined datastore values. The returned hash does
-  # not include default option values.
-  #
-  # @return [Hash<String, Object>] values explicitly defined on the data store which will override any default datastore values
-  attr_accessor :user_defined
-
-  #
-  # Was this entry actually set or just using its default
-  #
-  # @return [TrueClass, FalseClass]
-  def default?(key)
-    search_for(key).default?
-  end
-
-  #
-  # Clears the imported flag for the supplied key since it's being set
-  # directly.
-  #
-  def []=(k, v)
-    k = find_key_case(k)
-
-    opt = @options[k]
-    unless opt.nil?
-      if opt.validate_on_assignment?
-        unless opt.valid?(v, check_empty: false)
-          raise Msf::OptionValidateError.new(["Value '#{v}' is not valid for option '#{k}'"])
-        end
-        v = opt.normalize(v)
-      end
-    end
-
-    @user_defined[k] = v
-  end
-
-  #
-  # Case-insensitive wrapper around hash lookup
-  #
-  def [](k)
-    search_result = search_for(k)
-
-    search_result.value
-  end
-
-  #
-  # Case-insensitive wrapper around store; Skips option validation entirely
-  #
-  def store(k,v)
-    @user_defined[find_key_case(k)] = v
-  end
-
-  #
-  # Updates a value in the datastore with the specified name, k, to the
-  # specified value, v. Skips option validation entirely.
-  #
-  def update_value(k, v)
-    store(k, v)
-  end
-
-  #
-  # unset the current key from the datastore
-  # @param [String] key The key to search for
-  def unset(key)
-    k = find_key_case(key)
-    search_result = search_for(k)
-    @user_defined.delete(k)
-
-    search_result.value
-  end
-
-  # @deprecated use #{unset} instead, or set the value explicitly to nil
-  # @param [String] key The key to search for
-  def delete(key)
-    unset(key)
-  end
-
-  #
-  # Removes an option and any associated value
-  #
-  # @param [String] name the option name
-  # @return [nil]
-  def remove_option(name)
-    k = find_key_case(name)
-    @user_defined.delete(k)
-    @aliases.delete_if { |_, v| v.casecmp?(k) }
-    @options.delete_if { |option_name, _v| option_name.casecmp?(k) || option_name.casecmp?(name) }
-
-    nil
-  end
-
-  #
-  # This method is a helper method that imports the default value for
-  # all of the supplied options
-  #
-  def import_options(options, imported_by = nil, overwrite = true)
-    options.each_option do |name, option|
-      if self.options[name].nil? || overwrite
-        key = name
-        option.aliases.each do |a|
-          @aliases[a.downcase] = key.downcase
-        end
-        @options[key] = option
-      end
-    end
-  end
-
-  #
-  # Imports option values from a whitespace separated string in
-  # VAR=VAL format.
-  #
-  def import_options_from_s(option_str, delim = nil)
-    hash = {}
-
-    # Figure out the delimiter, default to space.
-    if (delim.nil?)
-      delim = /\s/
-
-      if (option_str.split('=').length <= 2 or option_str.index(',') != nil)
-        delim = ','
-      end
-    end
-
-    # Split on the delimiter
-    option_str.split(delim).each { |opt|
-      var, val = opt.split('=', 2)
-
-      next if (var =~ /^\s+$/)
-
-
-      # Invalid parse?  Raise an exception and let those bastards know.
-      if (var == nil or val == nil)
-        var = "unknown" if (!var)
-
-        raise Rex::ArgumentParseError, "Invalid option specified: #{var}",
-          caller
-      end
-
-      # Remove trailing whitespaces from the value
-      val.gsub!(/\s+$/, '')
-
-      # Store the value
-      hash[var] = val
-    }
-
-    merge!(hash)
-  end
-
-  #
-  # Imports values from a hash and stores them in the datastore.
-  #
-  # @deprecated use {#merge!} instead
-  # @return [nil]
-  def import_options_from_hash(option_hash, imported = true, imported_by = nil)
-    merge!(option_hash)
-  end
-
-  # Update defaults from a hash. These merged values are not validated by default.
-  #
-  # @param [Hash<String, Object>] hash The default values that should be used by the datastore
-  # @param [Object] imported_by Who imported the defaults, not currently used
-  # @return [nil]
-  def import_defaults_from_hash(hash, imported_by:)
-    @defaults.merge!(hash)
-  end
-
-  # TODO: Doesn't normalize data in the same vein as:
-  # https://github.com/rapid7/metasploit-framework/pull/6644
-  # @deprecated Use {#import_options}
-  def import_option(key, val, imported = true, imported_by = nil, option = nil)
-    store(key, val)
-
-    if option
-      option.aliases.each do |a|
-        @aliases[a.downcase] = key.downcase
-      end
-    end
-    @options[key] = option
-  end
-
-  # @return [Array<String>] The array of user defined datastore values, and registered option names
-  def keys
-    (@user_defined.keys + @options.keys).uniq(&:downcase)
-  end
-
-  # @return [Integer] The length of the registered keys
-  def length
-    keys.length
-  end
-
-  alias count length
-  alias size length
-
-  # @param [String] key
-  # @return [TrueClass, FalseClass] True if the key is present in the user defined values, or within registered options. False otherwise.
-  def key?(key)
-    matching_key = find_key_case(key)
-    keys.include?(matching_key)
-  end
-
-  alias has_key? key?
-  alias include? key?
-  alias member? key?
-
-  #
-  # Serializes the options in the datastore to a string.
-  #
-  def to_s(delim = ' ')
-    str = ''
-
-    keys.sort.each { |key|
-      str << "#{key}=#{self[key]}" + ((str.length) ? delim : '')
-    }
-
-    str
-  end
-
-  # Override Hash's to_h method so we can include the original case of each key
-  # (failing to do this breaks a number of places in framework and pro that use
-  # serialized datastores)
-  def to_h
-    datastore_hash = {}
-    self.keys.each do |k|
-      datastore_hash[k.to_s] = self[k].to_s
-    end
-    datastore_hash
-  end
-
-  # Hack on a hack for the external modules
-  def to_external_message_h
-    datastore_hash = {}
-
-    array_nester = ->(arr) do
-      if arr.first.is_a? Array
-        arr.map &array_nester
-      else
-        arr.map { |item| item.to_s.dup.force_encoding('UTF-8') }
-      end
-    end
-
-    self.keys.each do |k|
-      # TODO arbitrary depth
-      if self[k].is_a? Array
-        datastore_hash[k.to_s.dup.force_encoding('UTF-8')] = array_nester.call(self[k])
-      else
-        datastore_hash[k.to_s.dup.force_encoding('UTF-8')] = self[k].to_s.dup.force_encoding('UTF-8')
-      end
-    end
-    datastore_hash
-  end
-
-  #
-  # Persists the contents of the data store to a file
-  #
-  def to_file(path, name = 'global')
-    ini = Rex::Parser::Ini.new(path)
-
-    ini.add_group(name)
-
-    # Save all user-defined options to the file.
-    @user_defined.each_pair { |k, v|
-      ini[name][k] = v
-    }
-
-    ini.to_file(path)
-  end
-
-  #
-  # Imports datastore values from the specified file path using the supplied
-  # name
-  #
-  def from_file(path, name = 'global')
-    begin
-      ini = Rex::Parser::Ini.from_file(path)
-    rescue
-      return
-    end
-
-    if ini.group?(name)
-      merge!(ini[name])
-    end
-  end
-
-  #
-  # Return a copy of this datastore. Only string values will be duplicated, other values
-  # will share the same reference
-  # @return [Msf::DataStore] a new datastore instance
-  def copy
-    new_instance = self.class.new
-    new_instance.copy_state(self)
-    new_instance
-  end
-
-  #
-  # Merge the other object into the current datastore's aliases and imported hashes
-  #
-  # @param [Msf::Datastore, Hash] other
-  def merge!(other)
-    if other.is_a?(DataStoreWithFallbacks)
-      self.aliases.merge!(other.aliases)
-      self.options.merge!(other.options)
-      self.defaults.merge!(other.defaults)
-      other.user_defined.each do |k, v|
-        @user_defined[find_key_case(k)] = v
-      end
-    else
-      other.each do |k, v|
-        self.store(k, v)
-      end
-    end
-
-    self
-  end
-
-  alias update merge!
-
-  #
-  # Reverse Merge the other object into the current datastore's aliases and imported hashes
-  # Equivalent to ActiveSupport's reverse_merge! functionality.
-  #
-  # @param [Msf::Datastore] other
-  def reverse_merge!(other)
-    raise ArgumentError, "invalid error type #{other.class}, expected ::Msf::DataStore" unless other.is_a?(Msf::DataStoreWithFallbacks)
-
-    copy_state(other.merge(self))
-  end
-
-  #
-  # Override merge to ensure we merge the aliases and imported hashes
-  #
-  # @param [Msf::Datastore,Hash] other
-  def merge(other)
-    ds = self.copy
-    ds.merge!(other)
-  end
-
-  #
-  # Completely clear all values in the data store
-  #
-  def clear
-    self.options.clear
-    self.aliases.clear
-    self.defaults.clear
-    self.user_defined.clear
-
-    self
-  end
-
-  #
-  # Overrides the builtin 'each' operator to avoid the following exception on Ruby 1.9.2+
-  #    "can't add a new key into hash during iteration"
-  #
-  def each(&block)
-    list = []
-    self.keys.sort.each do |sidx|
-      list << [sidx, self[sidx]]
-    end
-    list.each(&block)
-  end
-
-  alias each_pair each
-
-  def each_key(&block)
-    self.keys.each(&block)
-  end
-
-  #
-  # Case-insensitive key lookup
-  #
-  # @return [String]
-  def find_key_case(k)
-    # Scan each alias looking for a key
-    search_k = k.downcase
-    if self.aliases.has_key?(search_k)
-      search_k = self.aliases[search_k]
-    end
-
-    # Check to see if we have an exact key match - otherwise we'll have to search manually to check case sensitivity
-    if @user_defined.key?(search_k) || options.key?(search_k)
-      return search_k
-    end
-
-    # Scan each key looking for a match
-    each_key do |rk|
-      if rk.casecmp(search_k) == 0
-        return rk
-      end
-    end
-
-    # Fall through to the non-existent value
-    k
-  end
-
-  # Search for a value within the current datastore, taking into consideration any registered aliases, fallbacks, etc.
-  #
-  # @param [String] key The key to search for
-  # @return [DataStoreSearchResult]
-  def search_for(key)
-    k = find_key_case(key)
-    return search_result(:user_defined, @user_defined[k]) if @user_defined.key?(k)
-
-    option = @options.fetch(k) { @options.find { |option_name, _option| option_name.casecmp?(k) }&.last }
-    if option
-      # If the key isn't present - check any additional fallbacks that have been registered with the option.
-      # i.e. handling the scenario of SMBUser not being explicitly set, but the option has registered a more
-      # generic 'Username' fallback
-      option.fallbacks.each do |fallback|
-      fallback_search = search_for(fallback)
-        if fallback_search.found?
-          return search_result(:option_fallback, fallback_search.value, fallback_key: fallback)
-        end
-      end
-    end
-
-    # Checking for imported default values, ignoring case again
-    imported_default_match = @defaults.find { |default_key, _default_value| default_key.casecmp?(k) }
-    return search_result(:imported_default, imported_default_match.last) if imported_default_match
-    return search_result(:option_default, option.default) if option
-
-    search_result(:not_found, nil)
-  end
-
-  protected
-
-  # These defaults will be used if the user has not explicitly defined a specific datastore value.
-  # These will be checked as a priority to any options that also provide defaults.
-  #
-  # @return [Hash{String => Msf::OptBase}] The hash of default values
-  attr_accessor :defaults
-
-  # @return [Hash{String => String}] The key is the old option name, the value is the new option name
-  attr_accessor :aliases
-
-  #
-  # Copy the state from the other Msf::DataStore. The state will be coped in a shallow fashion, other than
-  # imported and user_defined strings.
-  #
-  # @param [Msf::DataStore] other The other datastore to copy state from
-  # @return [Msf::DataStore] the current datastore instance
-  def copy_state(other)
-    self.options = other.options.dup
-    self.aliases = other.aliases.dup
-    self.defaults = other.defaults.transform_values { |value| value.kind_of?(String) ? value.dup : value }
-    self.user_defined = other.user_defined.transform_values { |value| value.kind_of?(String) ? value.dup : value }
-
-    self
-  end
-
-  # Raised when the specified key is not found
-  # @param [string] key
-  def key_error_for(key)
-    ::KeyError.new "key not found: #{key.inspect}"
-  end
-
-  #
-  # Simple dataclass for storing the result of a datastore search
-  #
-  class DataStoreSearchResult
-    # @return [String, nil] the key associated with the fallback value
-    attr_reader :fallback_key
-
-    # @return [object, nil] The value if found
-    attr_reader :value
-
-    def initialize(result, value, namespace: nil, fallback_key: nil)
-      @namespace = namespace
-      @result = result
-      @value = value
-      @fallback_key = fallback_key
-    end
-
-    def default?
-      result == :imported_default || result == :option_default || !found?
-    end
-
-    def found?
-      result != :not_found
-    end
-
-    def fallback?
-      result == :option_fallback
-    end
-
-    def global?
-      namespace == :global_data_store && found?
-    end
-
-  protected
-
-    # @return [Symbol] namespace Where the search result was found, i.e. a module datastore or global datastore
-    attr_reader :namespace
-
-    # @return [Symbol] result is one of `user_defined`, `not_found`, `option_fallback`, `option_default`, `imported_default`
-    attr_reader :result
-  end
-
-  def search_result(result, value, fallback_key: nil)
-    DataStoreSearchResult.new(result, value, namespace: :global_data_store, fallback_key: fallback_key)
-  end
-end
-
-end
diff --git a/lib/msf/core/feature_manager.rb b/lib/msf/core/feature_manager.rb
index 872087c35049..d822bb96dd98 100644
--- a/lib/msf/core/feature_manager.rb
+++ b/lib/msf/core/feature_manager.rb
@@ -15,7 +15,6 @@ class FeatureManager
 
     CONFIG_KEY = 'framework/features'
     WRAPPED_TABLES = 'wrapped_tables'
-    DATASTORE_FALLBACKS = 'datastore_fallbacks'
     FULLY_INTERACTIVE_SHELLS = 'fully_interactive_shells'
     MANAGER_COMMANDS = 'manager_commands'
     METASPLOIT_PAYLOAD_WARNINGS = 'metasploit_payload_warnings'
@@ -49,13 +48,6 @@ class FeatureManager
         default_value: false,
         developer_notes: 'Useful for developers, likely not to ever be useful for an average user'
       }.freeze,
-      {
-        name: DATASTORE_FALLBACKS,
-        description: 'When enabled you can consistently set username across modules, instead of setting SMBUser/FTPUser/BIND_DN/etc',
-        requires_restart: true,
-        default_value: true,
-        developer_notes: 'This functionality is enabled by default now, and the feature flag can be removed now'
-      }.freeze,
       {
         name: METASPLOIT_PAYLOAD_WARNINGS,
         description: 'When enabled Metasploit will output warnings about missing Metasploit payloads, for instance if they were removed by antivirus etc',
diff --git a/lib/msf/core/module.rb b/lib/msf/core/module.rb
index 749fd276a488..ff455d4716d1 100644
--- a/lib/msf/core/module.rb
+++ b/lib/msf/core/module.rb
@@ -136,7 +136,7 @@ def initialize(info = {})
       self.options.add_evasion_options(info['EvasionOptions'], self.class)
 
       # Create and initialize the data store for this module
-      self.datastore = ModuleDataStore.new(self)
+      self.datastore = Msf::ModuleDataStore.new(self)
 
       # Import default options into the datastore
       import_defaults
diff --git a/lib/msf/core/module/data_store.rb b/lib/msf/core/module/data_store.rb
index 675e400ab588..b4a231ba9b8a 100644
--- a/lib/msf/core/module/data_store.rb
+++ b/lib/msf/core/module/data_store.rb
@@ -21,7 +21,7 @@ def import_defaults(clear_datastore = true)
 
     # If there are default options, import their values into the datastore
     if (module_info['DefaultOptions'])
-      if datastore.is_a?(Msf::DataStoreWithFallbacks)
+      if datastore.is_a?(Msf::DataStore)
         self.datastore.import_defaults_from_hash(module_info['DefaultOptions'], imported_by: 'import_defaults')
       else
         self.datastore.import_options_from_hash(module_info['DefaultOptions'], true, 'self')
@@ -38,7 +38,7 @@ def import_defaults(clear_datastore = true)
   def import_target_defaults
     return unless defined?(targets) && targets && target && target.default_options
 
-    if self.datastore.is_a?(Msf::ModuleDataStoreWithFallbacks)
+    if self.datastore.is_a?(Msf::ModuleDataStore)
       datastore.import_defaults_from_hash(target.default_options, imported_by: 'import_target_defaults')
     else
       datastore.import_options_from_hash(target.default_options, true, 'self')
diff --git a/lib/msf/core/module/options.rb b/lib/msf/core/module/options.rb
index a7b3d0608369..23bc014db4c5 100644
--- a/lib/msf/core/module/options.rb
+++ b/lib/msf/core/module/options.rb
@@ -30,7 +30,7 @@ def validate
   def deregister_options(*names)
     names.each { |name|
       real_name = self.datastore.find_key_case(name)
-      if self.datastore.is_a?(Msf::DataStoreWithFallbacks)
+      if self.datastore.is_a?(Msf::DataStore)
         self.datastore.remove_option(name)
       else
         self.datastore.delete(name)
diff --git a/lib/msf/core/module_data_store.rb b/lib/msf/core/module_data_store.rb
index e846d4ad433f..609b2c5c7ace 100644
--- a/lib/msf/core/module_data_store.rb
+++ b/lib/msf/core/module_data_store.rb
@@ -10,20 +10,7 @@ module Msf
   ###
   class ModuleDataStore < DataStore
 
-    # Temporary forking logic for conditionally using the {Msf::ModuleDatastoreWithFallbacks} implementation.
-    #
-    # This method replaces the default `ModuleDataStore.new` with the ability to instantiate the `ModuleDataStoreWithFallbacks`
-    # class instead, if the feature is enabled
-    def self.new(m)
-      if Msf::FeatureManager.instance.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-        return Msf::ModuleDataStoreWithFallbacks.new(m)
-      end
-
-      instance = allocate
-      instance.send(:initialize, m)
-      instance
-    end
-
+    # @param [Msf::Module] m
     def initialize(m)
       super()
 
@@ -31,51 +18,63 @@ def initialize(m)
     end
 
     #
-    # Fetch the key from the local hash first, or from the framework datastore
-    # if we can't directly find it
-    #
-    def fetch(key)
-      key = find_key_case(key)
-      val = nil
-      val = super if(@imported_by[key] != 'self')
-      if (val.nil? and @_module and @_module.framework)
-        val = @_module.framework.datastore[key]
-      end
-      val = super if val.nil?
-      val
+    # Return a copy of this datastore. Only string values will be duplicated, other values
+    # will share the same reference
+    # @return [Msf::DataStore] a new datastore instance
+    def copy
+      new_instance = self.class.new(@_module)
+      new_instance.copy_state(self)
+      new_instance
     end
 
+    # Search for a value within the current datastore, taking into consideration any registered aliases, fallbacks, etc.
+    # If a value is not present in the current datastore, the global parent store will be referenced instead
     #
-    # Same as fetch
-    #
-    def [](key)
-      key = find_key_case(key)
-      val = nil
-      val = super if(@imported_by[key] != 'self')
-      if (val.nil? and @_module and @_module.framework)
-        val = @_module.framework.datastore[key]
+    # @param [String] key The key to search for
+    # @return [DataStoreSearchResult]
+    def search_for(key)
+      k = find_key_case(key)
+      return search_result(:user_defined, @user_defined[k]) if @user_defined.key?(k)
+
+      # Preference globally set values over a module's option default
+      framework_datastore_search = search_framework_datastore(key)
+      return framework_datastore_search if framework_datastore_search.found? && !framework_datastore_search.default?
+
+      option = @options.fetch(k) { @options.find { |option_name, _option| option_name.casecmp?(k) }&.last }
+      if option
+        # If the key isn't present - check any additional fallbacks that have been registered with the option.
+        # i.e. handling the scenario of SMBUser not being explicitly set, but the option has registered a more
+        # generic 'Username' fallback
+        option.fallbacks.each do |fallback|
+          fallback_search = search_for(fallback)
+          if fallback_search.found?
+            return search_result(:option_fallback, fallback_search.value, fallback_key: fallback)
+          end
+        end
       end
-      val = super if val.nil?
-      val
+
+      # Checking for imported default values, ignoring case again TODO: add Alias test for this
+      imported_default_match = @defaults.find { |default_key, _default_value| default_key.casecmp?(k) }
+      return search_result(:imported_default, imported_default_match.last) if imported_default_match
+      return search_result(:option_default, option.default) if option
+
+      search_framework_datastore(k)
     end
 
+    protected
+
+    # Search the framework datastore
     #
-    # Was this entry actually set or just using its default
-    #
-    def default?(key)
-      (@imported_by[key] == 'self')
+    # @param [String] key The key to search for
+    # @return [DataStoreSearchResult]
+    def search_framework_datastore(key)
+      return search_result(:not_found, nil) if @_module&.framework.nil?
+
+      @_module.framework.datastore.search_for(key)
     end
 
-    #
-    # Return a deep copy of this datastore.
-    #
-    def copy
-      ds = self.class.new(@_module)
-      self.keys.each do |k|
-        ds.import_option(k, self[k].kind_of?(String) ? self[k].dup : self[k], @imported[k], @imported_by[k])
-      end
-      ds.aliases = self.aliases.dup
-      ds
+    def search_result(result, value, fallback_key: nil)
+      DataStoreSearchResult.new(result, value, namespace: :module_data_store, fallback_key: fallback_key)
     end
   end
 end
diff --git a/lib/msf/core/module_data_store_with_fallbacks.rb b/lib/msf/core/module_data_store_with_fallbacks.rb
deleted file mode 100644
index 0109c74fbf78..000000000000
--- a/lib/msf/core/module_data_store_with_fallbacks.rb
+++ /dev/null
@@ -1,80 +0,0 @@
-# -*- coding: binary -*-
-module Msf
-
-  ###
-  #
-  # DataStore wrapper for modules that will attempt to back values against the
-  # framework's datastore if they aren't found in the module's datastore.  This
-  # is done to simulate global data store values.
-  #
-  ###
-  class ModuleDataStoreWithFallbacks < DataStoreWithFallbacks
-
-    # @param [Msf::Module] m
-    def initialize(m)
-      super()
-
-      @_module = m
-    end
-
-    #
-    # Return a copy of this datastore. Only string values will be duplicated, other values
-    # will share the same reference
-    # @return [Msf::DataStore] a new datastore instance
-    def copy
-      new_instance = self.class.new(@_module)
-      new_instance.copy_state(self)
-      new_instance
-    end
-
-    # Search for a value within the current datastore, taking into consideration any registered aliases, fallbacks, etc.
-    # If a value is not present in the current datastore, the global parent store will be referenced instead
-    #
-    # @param [String] key The key to search for
-    # @return [DataStoreSearchResult]
-    def search_for(key)
-      k = find_key_case(key)
-      return search_result(:user_defined, @user_defined[k]) if @user_defined.key?(k)
-
-      # Preference globally set values over a module's option default
-      framework_datastore_search = search_framework_datastore(key)
-      return framework_datastore_search if framework_datastore_search.found? && !framework_datastore_search.default?
-
-      option = @options.fetch(k) { @options.find { |option_name, _option| option_name.casecmp?(k) }&.last }
-      if option
-        # If the key isn't present - check any additional fallbacks that have been registered with the option.
-        # i.e. handling the scenario of SMBUser not being explicitly set, but the option has registered a more
-        # generic 'Username' fallback
-        option.fallbacks.each do |fallback|
-          fallback_search = search_for(fallback)
-          if fallback_search.found?
-            return search_result(:option_fallback, fallback_search.value, fallback_key: fallback)
-          end
-        end
-      end
-
-      # Checking for imported default values, ignoring case again TODO: add Alias test for this
-      imported_default_match = @defaults.find { |default_key, _default_value| default_key.casecmp?(k) }
-      return search_result(:imported_default, imported_default_match.last) if imported_default_match
-      return search_result(:option_default, option.default) if option
-
-      search_framework_datastore(k)
-    end
-
-    protected
-
-    # Search the framework datastore
-    #
-    # @param [String] key The key to search for
-    # @return [DataStoreSearchResult]
-    def search_framework_datastore(key)
-      return search_result(:not_found, nil) if @_module&.framework.nil?
-
-      @_module.framework.datastore.search_for(key)
-    end
-
-    def search_result(result, value, fallback_key: nil)
-      DataStoreSearchResult.new(result, value, namespace: :module_data_store, fallback_key: fallback_key)
-    end
-  end
-end
diff --git a/lib/msf/core/modules/external/templates/multi_scanner.erb b/lib/msf/core/modules/external/templates/multi_scanner.erb
index eebd6b210529..7dd5e936be96 100644
--- a/lib/msf/core/modules/external/templates/multi_scanner.erb
+++ b/lib/msf/core/modules/external/templates/multi_scanner.erb
@@ -27,7 +27,7 @@ class MetasploitModule < Msf::Auxiliary
 
   def run_batch(ips)
     datastore.delete('RHOSTS')
-    datastore.remove_option('RHOSTS') if self.datastore.is_a?(Msf::DataStoreWithFallbacks)
+    datastore.remove_option('RHOSTS') if self.datastore.is_a?(Msf::DataStore)
     datastore['rhosts'] = ips
 
     execute_module(<%= meta[:path] %>)
diff --git a/lib/msf/core/modules/external/templates/single_host_login_scanner.erb b/lib/msf/core/modules/external/templates/single_host_login_scanner.erb
index 39919ab1c3bf..bde198d3887d 100644
--- a/lib/msf/core/modules/external/templates/single_host_login_scanner.erb
+++ b/lib/msf/core/modules/external/templates/single_host_login_scanner.erb
@@ -24,7 +24,7 @@ class MetasploitModule < Msf::Auxiliary
   def run_host(ip)
     print_status("Running for #{ip}...")
     rhost = datastore.delete('RHOST')
-    datastore.remove_option('RHOST') if self.datastore.is_a?(Msf::DataStoreWithFallbacks)
+    datastore.remove_option('RHOST') if self.datastore.is_a?(Msf::DataStore)
     datastore['rhost'] = rhost
     datastore['userpass'] ||= build_credentials_array
     datastore['sleep_interval'] ||= userpass_interval
diff --git a/lib/msf/core/modules/external/templates/single_scanner.erb b/lib/msf/core/modules/external/templates/single_scanner.erb
index 30bed40dec1d..ae5a71985b1e 100644
--- a/lib/msf/core/modules/external/templates/single_scanner.erb
+++ b/lib/msf/core/modules/external/templates/single_scanner.erb
@@ -23,7 +23,7 @@ class MetasploitModule < Msf::Auxiliary
   def run_host(ip)
     print_status("Running for #{ip}...")
     rhost = datastore.delete('RHOST')
-    datastore.remove_option('RHOST') if self.datastore.is_a?(Msf::DataStoreWithFallbacks)
+    datastore.remove_option('RHOST') if self.datastore.is_a?(Msf::DataStore)
     datastore['rhost'] = rhost
     execute_module(<%= meta[:path] %>)
   end
diff --git a/lib/msf/core/option_group.rb b/lib/msf/core/option_group.rb
index a86fe94640f6..4db83403167b 100644
--- a/lib/msf/core/option_group.rb
+++ b/lib/msf/core/option_group.rb
@@ -36,7 +36,7 @@ def add_options(option_names)
     # Validates that any registered and required options are set
     #
     # @param options [Array<Msf::OptBase>] A modules registered options
-    # @param datastore [Msf::DataStore|Msf::DataStoreWithFallbacks] A modules datastore
+    # @param datastore [Msf::DataStore|Msf::DataStore] A modules datastore
     def validate(options, datastore)
       issues = {}
       required_options.each do |option_name|
diff --git a/lib/msf/core/payload.rb b/lib/msf/core/payload.rb
index 056e904ce911..42ec761fd741 100644
--- a/lib/msf/core/payload.rb
+++ b/lib/msf/core/payload.rb
@@ -475,7 +475,7 @@ def self.choose_payload(mod)
     lhost = mod.datastore['LHOST'] || Rex::Socket.source_address(mod.datastore['RHOST'] || '50.50.50.50')
 
     configure_payload = lambda do |payload|
-      if mod.datastore.is_a?(Msf::DataStoreWithFallbacks)
+      if mod.datastore.is_a?(Msf::DataStore)
         payload_defaults = { 'PAYLOAD' => payload }
 
         # Set LHOST if this is a reverse payload
diff --git a/lib/msf/core/rhosts_walker.rb b/lib/msf/core/rhosts_walker.rb
index b414abb214ff..f0997d1226a5 100644
--- a/lib/msf/core/rhosts_walker.rb
+++ b/lib/msf/core/rhosts_walker.rb
@@ -65,7 +65,7 @@ def each(&block)
       return unless block_given?
 
       parse(@value, @datastore).each do |result|
-        block.call(result) if result.is_a?(Msf::DataStore) || result.is_a?(Msf::DataStoreWithFallbacks)
+        block.call(result) if result.is_a?(Msf::DataStore)
       end
 
       nil
@@ -99,7 +99,7 @@ def errors(&block)
     # @return [Boolean] True if all items are valid, and there are at least some items present to iterate over. False otherwise.
     def valid?
       parsed_values = parse(@value, @datastore)
-      parsed_values.all? { |result| result.is_a?(Msf::DataStore) || result.is_a?(Msf::DataStoreWithFallbacks) } && parsed_values.count > 0
+      parsed_values.all? { |result| result.is_a?(Msf::DataStore) } && parsed_values.count > 0
     rescue StandardError => e
       elog('rhosts walker invalid', error: e)
       false
diff --git a/lib/msf/core/rpc/v10/rpc_core.rb b/lib/msf/core/rpc/v10/rpc_core.rb
index 5f9acab5729e..def9d8094bf4 100644
--- a/lib/msf/core/rpc/v10/rpc_core.rb
+++ b/lib/msf/core/rpc/v10/rpc_core.rb
@@ -64,7 +64,7 @@ def rpc_setg(var, val)
   # @example Here's how you would use this from the client:
   #  rpc.call('core.unsetg', 'MyGlobal')
   def rpc_unsetg(var)
-    if framework.datastore.is_a?(Msf::DataStoreWithFallbacks)
+    if framework.datastore.is_a?(Msf::DataStore)
       framework.datastore.unset(var)
     else
       framework.datastore.delete(var)
diff --git a/lib/msf/ui/console/command_dispatcher/core.rb b/lib/msf/ui/console/command_dispatcher/core.rb
index 5599879afad6..27dfb644d0e0 100644
--- a/lib/msf/ui/console/command_dispatcher/core.rb
+++ b/lib/msf/ui/console/command_dispatcher/core.rb
@@ -2081,7 +2081,7 @@ def cmd_set_help
     print_line "datastore.  Use -g to operate on the global datastore."
     print_line
     print_line "If setting a PAYLOAD, this command can take an index from `show payloads'."
-    print @@set_opts.usage if framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
+    print @@set_opts.usage
     print_line
   end
 
@@ -2103,7 +2103,7 @@ def cmd_set(*args)
       elsif args[0] == '-a'
         args.shift
         append = true
-      elsif (args[0] == '-c' || args[0] == '--clear') && framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
+      elsif (args[0] == '-c' || args[0] == '--clear')
         args.shift
         clear = true
       else
@@ -2271,7 +2271,7 @@ def cmd_setg_help
     print_line "Usage: setg [option] [value]"
     print_line
     print_line "Exactly like set -g, set a value in the global datastore."
-    print @@setg_opts.usage if framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
+    print @@setg_opts.usage
     print_line
   end
 
@@ -2433,83 +2433,18 @@ def cmd_getg_tabs(str, words)
   end
 
   def cmd_unset_help
-    if framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-      print_line "Usage: unset [-g] var1 var2 var3 ..."
-      print_line
-      print_line "The unset command is used to unset one or more variables."
-      print_line "To flush all entries, specify 'all' as the variable name."
-      print_line "With -g, operates on global datastore variables."
-      print_line
-    else
-      print_line "Usage: unset [options] var1 var2 var3 ..."
-      print_line
-      print_line "The unset command is used to unset one or more variables which have been set by the user."
-      print_line "To update all entries, specify 'all' as the variable name."
-      print @@unset_opts.usage
-      print_line
-    end
+    print_line "Usage: unset [-g] var1 var2 var3 ..."
+    print_line
+    print_line "The unset command is used to unset one or more variables."
+    print_line "To flush all entries, specify 'all' as the variable name."
+    print_line "With -g, operates on global datastore variables."
+    print_line
   end
 
   #
   # Unsets a value if it's been set.
   #
   def cmd_unset(*args)
-    if framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-      return cmd_unset_with_fallbacks(*args)
-    end
-
-    # Figure out if these are global variables
-    global = false
-
-    if (args[0] == '-g')
-      args.shift
-      global = true
-    end
-
-    # Determine which data store we're operating on
-    if (active_module and global == false)
-      datastore = active_module.datastore
-    else
-      datastore = framework.datastore
-    end
-
-    # No arguments?  No cookie.
-    if (args.length == 0)
-      cmd_unset_help
-      return false
-    end
-
-    # If all was specified, then flush all of the entries
-    if args[0] == 'all'
-      print_line("Flushing datastore...")
-
-      # Re-import default options into the module's datastore
-      if (active_module and global == false)
-        active_module.import_defaults
-      # Or simply clear the global datastore
-      else
-        datastore.clear
-      end
-
-      return true
-    end
-
-    while ((val = args.shift))
-      if (driver.on_variable_unset(global, val) == false)
-        print_error("The variable #{val} cannot be unset at this time.")
-        next
-      end
-
-      print_line("Unsetting #{val}...")
-
-      datastore.delete(val)
-    end
-  end
-
-  #
-  # Unsets a value if it's been set, resetting the value back to a default value
-  #
-  def cmd_unset_with_fallbacks(*args)
     if args.include?('-h') || args.include?('--help')
       cmd_unset_help
       return
@@ -2591,7 +2526,7 @@ def cmd_unsetg_help
     print_line "Usage: unsetg [options] var1 var2 var3 ..."
     print_line
     print_line "Exactly like unset -g, unset global variables, or all"
-    print @@unsetg_opts.usage if framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
+    print @@unsetg_opts.usage
     print_line
   end
 
diff --git a/lib/msf/ui/console/driver.rb b/lib/msf/ui/console/driver.rb
index bc42cb1a4cb9..3dc33211908e 100644
--- a/lib/msf/ui/console/driver.rb
+++ b/lib/msf/ui/console/driver.rb
@@ -611,10 +611,6 @@ def handle_payload(val)
       return false
     elsif active_module && (active_module.exploit? || active_module.evasion?)
       return false unless active_module.is_payload_compatible?(val)
-    elsif active_module && !framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-      active_module.datastore.clear_non_user_defined
-    elsif framework && !framework.features.enabled?(Msf::FeatureManager::DATASTORE_FALLBACKS)
-      framework.datastore.clear_non_user_defined
     end
   end
 
diff --git a/lib/msf/ui/console/module_option_tab_completion.rb b/lib/msf/ui/console/module_option_tab_completion.rb
index f8ecd07216e3..578f5f23630b 100644
--- a/lib/msf/ui/console/module_option_tab_completion.rb
+++ b/lib/msf/ui/console/module_option_tab_completion.rb
@@ -17,10 +17,10 @@ module ModuleOptionTabCompletion
         #   stage since the command itself has been completed.
         def tab_complete_datastore_names(datastore, _str, _words)
           keys = (
-            Msf::DataStoreWithFallbacks::GLOBAL_KEYS +
+            Msf::DataStore::GLOBAL_KEYS +
               datastore.keys
           )
-          keys.concat(datastore.options.values.flat_map(&:fallbacks)) if datastore.is_a?(Msf::DataStoreWithFallbacks)
+          keys.concat(datastore.options.values.flat_map(&:fallbacks)) if datastore.is_a?(Msf::DataStore)
           keys.uniq! { |key| key.downcase }
           keys
         end
diff --git a/lib/rex/post/meterpreter/ui/console/command_dispatcher/lanattacks/dhcp.rb b/lib/rex/post/meterpreter/ui/console/command_dispatcher/lanattacks/dhcp.rb
index b58f4dfb7010..46d06a76e07a 100644
--- a/lib/rex/post/meterpreter/ui/console/command_dispatcher/lanattacks/dhcp.rb
+++ b/lib/rex/post/meterpreter/ui/console/command_dispatcher/lanattacks/dhcp.rb
@@ -189,7 +189,7 @@ def cmd_dhcp_load_options(*args)
 
     datastore = args.shift
 
-    unless datastore.is_a?(Hash) || datastore.is_a?(Msf::DataStoreWithFallbacks)
+    unless datastore.is_a?(Hash) || datastore.is_a?(Msf::DataStore)
       print_dhcp_load_options_usage
       return true
     end
diff --git a/spec/lib/msf/core/data_store_with_fallbacks_spec.rb b/spec/lib/msf/core/data_store_with_fallbacks_spec.rb
index e6c0d2bfe2bf..eab0a36f4646 100644
--- a/spec/lib/msf/core/data_store_with_fallbacks_spec.rb
+++ b/spec/lib/msf/core/data_store_with_fallbacks_spec.rb
@@ -700,7 +700,7 @@
   end
 end
 
-RSpec.describe Msf::DataStoreWithFallbacks do
+RSpec.describe Msf::DataStore do
   include_context 'datastore subjects'
 
   subject(:default_subject) do
@@ -712,11 +712,11 @@
   it_behaves_like 'a datastore'
 end
 
-RSpec.describe Msf::ModuleDataStoreWithFallbacks do
+RSpec.describe Msf::ModuleDataStore do
   include_context 'datastore subjects'
 
   let(:framework_datastore) do
-    Msf::DataStoreWithFallbacks.new
+    Msf::DataStore.new
   end
   let(:mod) do
     framework = instance_double(Msf::Framework, datastore: framework_datastore)
diff --git a/spec/lib/msf/core/option_group_spec.rb b/spec/lib/msf/core/option_group_spec.rb
index d6669b942974..2980cdd6a5a1 100644
--- a/spec/lib/msf/core/option_group_spec.rb
+++ b/spec/lib/msf/core/option_group_spec.rb
@@ -53,7 +53,7 @@
     let(:option_names) { ['not_required_name', required_option_name] }
     let(:required_names) { [required_option_name] }
     let(:options) { instance_double(Msf::OptionContainer) }
-    let(:datastore) { instance_double(Msf::DataStoreWithFallbacks) }
+    let(:datastore) { instance_double(Msf::DataStore) }
 
     context 'when there are no required options' do
       subject { described_class.new(name: 'name', description: 'description', option_names: option_names) }
diff --git a/spec/modules/auxiliary/scanner/ssh/ssh_login_spec.rb b/spec/modules/auxiliary/scanner/ssh/ssh_login_spec.rb
index 854019788083..31f454527155 100644
--- a/spec/modules/auxiliary/scanner/ssh/ssh_login_spec.rb
+++ b/spec/modules/auxiliary/scanner/ssh/ssh_login_spec.rb
@@ -24,7 +24,7 @@
     let(:credential) do
       Metasploit::Framework::Credential.new(private: password, public: username)
     end
-    let(:datastore) { Msf::ModuleDataStoreWithFallbacks.new(subject) }
+    let(:datastore) { Msf::ModuleDataStore.new(subject) }
     let(:host) { '10.10.10.10' }
     let(:module_manager) { instance_double(Msf::ModuleManager) }
     let(:password) { 'secret' }
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index 6e0463ca4672..090cdc799f80 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -159,12 +159,6 @@
     end
   end
 
-  if ENV['MSF_FEATURE_DATASTORE_FALLBACKS']
-    config.before(:suite) do
-      Msf::FeatureManager.instance.set(Msf::FeatureManager::DATASTORE_FALLBACKS, true)
-    end
-  end
-
   if ENV['MSF_FEATURE_DEFER_MODULE_LOADS']
     config.before(:suite) do
       Msf::FeatureManager.instance.set(Msf::FeatureManager::DEFER_MODULE_LOADS, true)
diff --git a/spec/support/shared/examples/msf/core/optional_session.rb b/spec/support/shared/examples/msf/core/optional_session.rb
index c1a1080f21b0..a6f61122c239 100644
--- a/spec/support/shared/examples/msf/core/optional_session.rb
+++ b/spec/support/shared/examples/msf/core/optional_session.rb
@@ -7,7 +7,7 @@
   include_context 'Msf::Simple::Framework'
 
   let(:options) { instance_double(Msf::OptionContainer) }
-  let(:datastore) { instance_double(Msf::DataStoreWithFallbacks) }
+  let(:datastore) { instance_double(Msf::DataStore) }
   let(:session) { instance_double(Msf::Sessions::SMB) }
   let(:session_group) { instance_double(Msf::OptionGroup) }
   let(:rhost_group) { instance_double(Msf::OptionGroup) }
diff --git a/spec/support/shared/examples/msf/module/data_store.rb b/spec/support/shared/examples/msf/module/data_store.rb
index c51388cc89c4..10e15cced701 100644
--- a/spec/support/shared/examples/msf/module/data_store.rb
+++ b/spec/support/shared/examples/msf/module/data_store.rb
@@ -2,4 +2,4 @@
   it { is_expected.to respond_to :datastore }
   it { is_expected.to respond_to :import_defaults }
   it { is_expected.to respond_to :share_datastore }
-end
\ No newline at end of file
+end