diff --git a/lib/ldclient-rb/impl/file_data.rb b/lib/ldclient-rb/impl/file_data.rb new file mode 100644 index 00000000..eaa8b13b --- /dev/null +++ b/lib/ldclient-rb/impl/file_data.rb @@ -0,0 +1,7 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/file_data/document" +require "ldclient-rb/impl/file_data/merge" +require "ldclient-rb/impl/file_data/poller" +require "ldclient-rb/impl/file_data/reloader" +require "ldclient-rb/impl/file_data/watcher" diff --git a/lib/ldclient-rb/impl/file_data/document.rb b/lib/ldclient-rb/impl/file_data/document.rb new file mode 100644 index 00000000..fa02cfba --- /dev/null +++ b/lib/ldclient-rb/impl/file_data/document.rb @@ -0,0 +1,267 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/model/serialization" + +require "json" +require "yaml" + +module LaunchDarkly + module Impl + # + # Shared file reading, parsing, and merging code for the components that load flag and + # segment data from local files. + # + # @private + # + module FileData + # + # Raised when a file cannot be read or parsed. It carries the path so that callers can + # tell a per-file failure from a failure to merge the files' contents. + # + class ReadError < StandardError + # @return [String] + attr_reader :path + + # @return [Boolean] true when the file does not exist + attr_reader :missing + + # + # @param path [String] + # @param message [String] + # @param missing [Boolean] + # + def initialize(path, message, missing: false) + super("#{message} [#{path}]") + @path = path + @missing = missing + end + end + + # + # The parsed form of one data file. A document may contain full flag definitions, flag key + # to value entries, and segment definitions. Every hash has symbol keys. + # + class Document + # @return [Hash{Symbol => Hash}] + attr_reader :flags + + # @return [Hash{Symbol => Object}] + attr_reader :flag_values + + # @return [Hash{Symbol => Hash}] + attr_reader :segments + + # + # @param flags [Hash{Symbol => Hash}] + # @param flag_values [Hash{Symbol => Object}] + # @param segments [Hash{Symbol => Hash}] + # + def initialize(flags: {}, flag_values: {}, segments: {}) + @flags = flags + @flag_values = flag_values + @segments = segments + end + + # The document of YAML content that is only a null, such as an empty file, a comment, or + # "---": a document with no entries. + NONE = new(flags: {}.freeze, flag_values: {}.freeze, segments: {}.freeze).freeze + + # + # Parses the content of a data file in the given format. {FileData.format_of} chooses the + # format of a file from its name. + # + # Whether content with no entries is valid depends on the format, like the rest of the + # content. Content that is empty or only white space is not a JSON document, so it fails to + # parse. In YAML it is a document with no entries, {NONE}, and so is a YAML document that is + # only a null, such as one with only a comment. A document that is not a mapping, or whose + # "flags", "flagValues", or "segments" member is not a mapping, is an error, and so is + # content that is not valid UTF-8, in either format. + # + # @param content [String] UTF-8 content + # @param format [Symbol] `:json` or `:yaml` + # @return [Document] + # @raise [ArgumentError] if the content is not a valid document + # @raise [JSON::ParserError, Psych::SyntaxError] if the content cannot be parsed + # + def self.parse(content, format:) + FileData.check_encoding(content) + case format + when :json + # A file that is being rewritten in place can be empty for a moment. Failing to parse + # keeps its last good data until it is complete. + raise ArgumentError, "the file has no content, which is not a JSON document" if FileData.blank?(content) + + raw = JSON.parse(content) + when :yaml + # The YAML parser rejects some white space, such as a tab, that is still no content. + return NONE if FileData.blank?(content) + + raw = YAML.safe_load(content) + return NONE if raw.nil? + else + raise ArgumentError, "unknown file format #{format.inspect}" + end + raise ArgumentError, "file content must be an object" unless raw.is_a?(Hash) + + data = FileData.symbolize_keys(raw) + Document.new( + flags: section(data, :flags), + flag_values: section(data, :flagValues), + segments: section(data, :segments) + ) + end + + # + # Reads and parses one data file. + # + # @param path [String] + # @return [Document] + # @raise [ReadError] if the file cannot be read or parsed + # + def self.read(path) + content = FileData.read_file(path) + FileData.parse_file(path, content) + end + + private_class_method def self.section(data, name) + value = data[name] + return {} if value.nil? + raise ArgumentError, "\"#{name}\" must be an object" unless value.is_a?(Hash) + + value + end + end + + # The endings of the names of the files that are read as YAML, compared without regard to + # case. Every other file is read as JSON. + YAML_FILE_NAME_ENDINGS = [".yaml", ".yml"].freeze + + # The bytes that count as white space when deciding whether content is empty. + WHITE_SPACE_BYTES = [0x20, 0x09, 0x0a, 0x0d, 0x0b, 0x0c].freeze + private_constant :WHITE_SPACE_BYTES + + # + # Returns the format of a data file, which its name decides: a name that ends in ".yaml" or + # ".yml", in any case, is YAML, and any other name is JSON. + # + # @param path [String] + # @return [Symbol] `:json` or `:yaml` + # + def self.format_of(path) + path.downcase.end_with?(*YAML_FILE_NAME_ENDINGS) ? :yaml : :json + end + + # + # Raises an error if the content is not valid UTF-8. + # + # @param content [String] + # @raise [ArgumentError] if the content is not valid UTF-8 + # + def self.check_encoding(content) + return if content.valid_encoding? + + if content.b.start_with?(*UTF16_AND_UTF32_BOMS) + raise ArgumentError, "the file is encoded in UTF-16 or UTF-32; data files must be UTF-8" + end + raise ArgumentError, "the file is not valid UTF-8" + end + + # + # Returns true if the content is empty or only white space. + # + # @param content [String] + # @return [Boolean] + # + def self.blank?(content) + content.each_byte.all? { |byte| WHITE_SPACE_BYTES.include?(byte) } + end + + # The byte order marks of UTF-8 and of the encodings that files must not use. + UTF8_BOM = "\xEF\xBB\xBF".b.freeze + UTF16_AND_UTF32_BOMS = ["\xFF\xFE".b, "\xFE\xFF".b, "\x00\x00\xFE\xFF".b].freeze + private_constant :UTF8_BOM, :UTF16_AND_UTF32_BOMS + + # + # Reads the raw content of one file, as UTF-8 and without a UTF-8 byte order mark, whatever + # the process's default encoding. The bytes are not checked here: {Document.parse} rejects + # content that is not valid UTF-8, so that such a file fails to load like any other content + # that cannot be parsed. + # + # @param path [String] + # @return [String] + # @raise [ReadError] if the file cannot be read + # + def self.read_file(path) + content = File.binread(path) + content = content.byteslice(UTF8_BOM.bytesize..) if content.start_with?(UTF8_BOM) + content.force_encoding(Encoding::UTF_8) + rescue Errno::ENOENT => e + raise ReadError.new(path, "unable to read file: #{e.message}", missing: true) + rescue SystemCallError, IOError => e + raise ReadError.new(path, "unable to read file: #{e.message}") + end + + # + # Parses raw content that was read from the given path. + # + # @param path [String] + # @param content [String] + # @return [Document] + # @raise [ReadError] if the content cannot be parsed + # + def self.parse_file(path, content) + Document.parse(content, format: format_of(path)) + rescue StandardError => e + raise ReadError.new(path, "error parsing file: #{e.message}") + end + + # + # Recursively converts hash keys to symbols. The SDK expects all data model objects to + # have symbol keys. + # + # @param value [Object] + # @return [Object] + # + def self.symbolize_keys(value) + case value + when Hash + value.to_h { |k, v| [k.to_s.to_sym, symbolize_keys(v)] } + when Array + value.map { |v| symbolize_keys(v) } + else + value + end + end + + # + # Expands a flag key to value entry into a full flag definition that returns the given + # value for every context. The flag is on, has the value as its only variation, and + # serves that variation as its fallthrough, so an evaluation reports the FALLTHROUGH + # reason kind. + # + # @param key [String] + # @param value [Object] + # @return [Hash] + # + def self.make_flag_with_value(key, value) + { + key: key, + on: true, + version: 1, + fallthrough: { variation: 0 }, + variations: [value], + } + end + + # + # Converts each path to an absolute path. + # + # @param paths [Array, String] + # @return [Array] + # + def self.absolute_paths(paths) + Array(paths).map { |p| File.absolute_path(p.to_s) } + end + end + end +end diff --git a/lib/ldclient-rb/impl/file_data/merge.rb b/lib/ldclient-rb/impl/file_data/merge.rb new file mode 100644 index 00000000..270f84a3 --- /dev/null +++ b/lib/ldclient-rb/impl/file_data/merge.rb @@ -0,0 +1,157 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/data_store" +require "ldclient-rb/impl/file_data/document" +require "ldclient-rb/impl/model/serialization" + +module LaunchDarkly + module Impl + module FileData + # + # Values for the duplicate keys handling option. They select what happens when the same + # flag or segment key appears in more than one document. + # + module DuplicateKeysHandling + # A duplicated key makes the merge fail. + FAIL = :fail + + # Only the first occurrence of a duplicated key is kept, in the order the documents were given. + IGNORE = :ignore + + ALL = [FAIL, IGNORE].freeze + end + + # + # Raised when documents cannot be combined, for example because a key is duplicated, an + # entry is not an object, or an entry cannot be deserialized into the data model. + # + class MergeError < StandardError + end + + # + # Counts the entries the merge kept from one document. + # + DocumentSummary = Struct.new(:flags, :segments) + + # + # Describes one configured file after a reload. `present` is false when the file does not + # exist and missing files are skipped. + # + FileSummary = Struct.new(:path, :present, :flags, :segments) + + # + # The merged items from one or more documents. + # + class MergeResult + # @return [Hash{Symbol => LaunchDarkly::Impl::Model::FeatureFlag}] + attr_reader :flags + + # @return [Hash{Symbol => LaunchDarkly::Impl::Model::Segment}] + attr_reader :segments + + # @return [Array] one entry per input document, in order + attr_reader :documents + + # @return [Array] set by the Reloader, one entry per configured file, in order + attr_accessor :files + + def initialize(flags, segments, documents) + @flags = flags + @segments = segments + @documents = documents + @files = [] + end + + # @return [Boolean] + def empty? + @flags.empty? && @segments.empty? + end + end + + # + # Combines the items of the given documents into one set of flags and one set of segments. + # Flag key to value entries expand into full flag definitions. Entries are deserialized into + # the SDK's data model classes, which validate them. The documents are processed in order, + # and the configured duplicate keys handling applies when the same key appears more than once. + # + # Items are keyed by the key under which they appear in the document. An entry that has no + # "key" member receives that key. A missing "version" defaults to 1. + # + # @param documents [Array] + # @param duplicate_keys_handling [Symbol] one of the {DuplicateKeysHandling} values + # @param logger [Logger, nil] receives data model validation messages + # @return [MergeResult] + # @raise [MergeError] if the documents cannot be combined + # + def self.merge(documents, duplicate_keys_handling: DuplicateKeysHandling::FAIL, logger: nil) + flags = {} + segments = {} + summaries = [] + + documents.each do |document| + summary = DocumentSummary.new(0, 0) + + document.flags.each do |key, data| + data = prepare_entry("flag", key, data) + item = deserialize(DataStore::FEATURES, "flag", key, data, logger) + summary.flags += 1 if insert(flags, "flag", key, item, duplicate_keys_handling) + end + + document.flag_values.each do |key, value| + data = make_flag_with_value(key.to_s, value) + item = Model.deserialize(DataStore::FEATURES, data, logger) + summary.flags += 1 if insert(flags, "flag", key, item, duplicate_keys_handling) + end + + document.segments.each do |key, data| + data = prepare_entry("segment", key, data) + item = deserialize(DataStore::SEGMENTS, "segment", key, data, logger) + summary.segments += 1 if insert(segments, "segment", key, item, duplicate_keys_handling) + end + + summaries << summary + end + + MergeResult.new(flags, segments, summaries) + end + + # + # Validates one full flag or segment entry and fills in its key and version. + # + private_class_method def self.prepare_entry(category, key, data) + raise MergeError, "#{category} \"#{key}\" is not an object" unless data.is_a?(Hash) + + data = data.dup + data[:key] = key.to_s if data[:key].nil? + data[:version] = 1 if data[:version].nil? + data + end + + # + # Deserializes one entry into its data model class. The model classes raise for a value + # that does not fit the schema, and that failure is reported as a merge error that names + # the entry. + # + private_class_method def self.deserialize(kind, category, key, data, logger) + Model.deserialize(kind, data, logger) + rescue => e + raise MergeError, "#{category} \"#{key}\" is not valid: #{e.message}" + end + + # + # Adds an item unless its key was already seen. Returns true when it added the item. + # + private_class_method def self.insert(items, category, key, item, duplicate_keys_handling) + key = key.to_sym + if items.key?(key) + return false if duplicate_keys_handling == DuplicateKeysHandling::IGNORE + + raise MergeError, "#{category} key \"#{key}\" was used more than once" + end + + items[key] = item + true + end + end + end +end diff --git a/lib/ldclient-rb/impl/file_data/poller.rb b/lib/ldclient-rb/impl/file_data/poller.rb new file mode 100644 index 00000000..73d9db5e --- /dev/null +++ b/lib/ldclient-rb/impl/file_data/poller.rb @@ -0,0 +1,87 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/repeating_task" + +module LaunchDarkly + module Impl + module FileData + # + # Detects changes to a set of files by examining them on a fixed interval. Use it where file + # system change notifications are not available or not reliable. A change to the + # modification time or the size of any file invokes the callback. A file that appears or + # disappears is also a change. A file that cannot be examined counts as absent. + # + # The poller samples the files once per interval and compares only modification time and + # size. A rewrite that keeps both values is not detected. + # + # Detection is generous. The callback can run for a change that does not alter the + # effective data. Feed it into a {Reloader}, whose debouncing and skip-unchanged handling + # absorb the excess. + # + # @private + # + class Poller + # The observed state of one file, or its absence. + FileState = Struct.new(:exists, :mtime, :size) + + ABSENT = FileState.new(false, nil, nil).freeze + private_constant :ABSENT + + # + # Creates and starts a poller. It examines the files once before it returns, so only later + # changes invoke the callback. Call {#stop} to stop it. + # + # @param paths [Array] absolute paths of the files to examine + # @param interval [Numeric] seconds between examinations + # @param on_change [#call] invoked with no arguments when a change is detected + # @param logger [Logger] + # + def initialize(paths, interval, on_change, logger) + @paths = paths + @on_change = on_change + @last = Poller.observe_all(paths) + @task = RepeatingTask.new(interval, interval, method(:examine), logger, "LD/FileDataPoller") + @task.start + end + + # + # Stops the poller and waits for the worker thread to finish. A callback that is already + # running completes first. + # + def stop + @task.stop + end + + # + # Examines every file and returns the observed states, in path order. + # + # @param paths [Array] + # @return [Array] + # + def self.observe_all(paths) + paths.map { |path| observe(path) } + end + + # + # Examines one file. The state is that of the file a symbolic link leads to. + # + # @param path [String] + # @return [FileState] + # + def self.observe(path) + stat = File.stat(path) + FileState.new(true, stat.mtime, stat.size) + rescue SystemCallError + ABSENT + end + + private def examine + current = Poller.observe_all(@paths) + changed = current != @last + @last = current + @on_change.call if changed + end + end + end + end +end diff --git a/lib/ldclient-rb/impl/file_data/reloader.rb b/lib/ldclient-rb/impl/file_data/reloader.rb new file mode 100644 index 00000000..224f5b9f --- /dev/null +++ b/lib/ldclient-rb/impl/file_data/reloader.rb @@ -0,0 +1,562 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/file_data/document" +require "ldclient-rb/impl/file_data/merge" +require "ldclient-rb/impl/file_data/poller" +require "ldclient-rb/impl/util" + +require "digest" + +module LaunchDarkly + module Impl + module FileData + # + # Owns the reload cycle for a set of data files. It serializes reloads, debounces change + # signals, retains the last good result on failure (by not calling `apply`), retries after + # failures, holds back a file that was emptied before applying it, and skips no-op + # applications. + # + # Applications and editors rewrite files in place: they truncate the file, then write the new + # content. A load in between finds the file empty, or, for YAML, can find only a comment or + # "---", which the YAML parser reads as nothing. Whether a file with no entries is valid + # depends on its format (see {Document.parse}), so a parser cannot tell an emptied file from + # one in the middle of a rewrite. So a load that finds a file with no content (empty, white + # space, or YAML that is only a null), while the data currently applied has content from that + # file, holds the file back: it reads the other files, so that files emptied together are + # held back together, then keeps the current data and arms the retry, without reporting a + # failure. A file that cannot be read or parsed still fails the load. A YAML document such as + # `{}` is content, and is applied at once. + # + # A file is held back until it has stayed so, without being written, for the retry delay, + # counted from the load that read it. A change of its state (its modification time and size, + # read through any symbolic link) means it was written since a load last found it so, for + # example rewritten and truncated again between two loads, so the delay starts over. Each file + # is held back for at most five retry delays from the first load that held it back. The bound + # goes on across reads that find the file with content, and starts over only at a load that + # holds nothing back, whether that load applies, is skipped as unchanged, or fails, for each + # file it read. A load reads every file, even after one fails to read or parse, and fails with + # the first failure after them. So a file that keeps being rewritten, or files rewritten in + # turn, cannot hold the data back without end. A read that finds a file with content, even content + # that fails to parse, or finds it missing, ends its empty state, so its next emptying starts + # a new delay. The retry that a hold arms comes when the first held file's + # delay or bound is over, plus a tenth of the retry delay. A file that contributed no content + # to the data currently applied, such as a placeholder that has been empty from the start, a + # file that was missing, or a file whose emptying has been applied, is parsed at once. So is every file when there are no + # automatic retries, because nothing would read it again. After the delay, an empty JSON file + # fails to parse and the data stays; an empty YAML file is a document with no entries. + # + # The worker thread starts on the first {#reload_now} or {#trigger} call rather than in the + # constructor. A reloader can be constructed by a component whose lifecycle never uses it, + # and construction alone must not leak a thread. + # + # @private + # + class Reloader + # A settle window long enough to coalesce the burst of change notifications produced by a + # single file edit, and short enough to stay responsive. In seconds. + DEFAULT_DEBOUNCE_DELAY = 0.1 + + # Bounds how long a failed reload can go uncorrected when no further change notification + # arrives, for example when the failure came from reading a file mid-write. Reading a local + # file is cheap, so this can be short. In seconds. + DEFAULT_RETRY_DELAY = 1.0 + + # A file is held back for at most this many retry delays from the first load that held it + # back. The bound starts over at a load that holds nothing back. + MAX_HOLD_RETRY_DELAYS = 5 + + # A file whose content is in effect and that loads have found with no content: since when it + # has been so without being written, or nil once a load read it with content or found it + # missing; its {Poller::FileState} then; and when a load first held it back, or nil. + EmptyFile = Struct.new(:since, :state, :first_held) + private_constant :EmptyFile + + # + # @param paths [Array] the files to load, already resolved to absolute paths. The + # order is significant: it determines which file wins under the duplicate keys handling. + # @param logger [Logger] + # @param apply [#call] invoked with each successfully merged {MergeResult}. Calls are + # serialized, so implementations do not need their own synchronization against other + # reloads. An exception from `apply` is a failed reload: it is reported through + # `on_error`, the result is not remembered as applied, and the reload is retried. + # `apply` and `on_error` must not call back into {#stop}. + # @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. + # @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. + # When false, a missing file fails the reload like any other read error. + # @param debounce_delay [Numeric] seconds to wait after a {#trigger} call for further calls + # to settle before reloading. If zero or negative, each trigger reloads at once. + # @param retry_delay [Numeric] seconds to wait after a failed reload before retrying + # automatically. A file whose content is in effect and that is found empty is held back + # for the same delay. If zero or negative, there is no automatic retry, and an empty file + # is parsed as it is. + # @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 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. + # + def initialize(paths:, logger:, apply:, on_error: nil, + duplicate_keys_handling: DuplicateKeysHandling::FAIL, + skip_missing_paths: false, + debounce_delay: DEFAULT_DEBOUNCE_DELAY, + retry_delay: DEFAULT_RETRY_DELAY, + skip_unchanged: false, + clock: nil) + @paths = paths + @logger = logger + @apply = apply + @on_error = on_error + @duplicate_keys_handling = duplicate_keys_handling + @skip_missing_paths = skip_missing_paths + @debounce_delay = debounce_delay + @retry_delay = retry_delay + @skip_unchanged = skip_unchanged + @clock = clock || method(:monotonic_now) + + # Guards the scheduling state below and wakes the worker. + @mutex = Mutex.new + @cond = ConditionVariable.new + @worker = nil + @stopped = false + @trigger_pending = false + @requested_retry_delay = nil + @debounce_deadline = nil + @retry_deadline = nil + + # Serializes the actual load work between reload_now and the worker. + @reload_mutex = Mutex.new + @last_good_digest = nil + @last_error_message = nil + # Whether each configured path contributed content to the data currently applied; the + # EmptyFile of each path whose content is in effect and that loads have found with no + # content since; whether the last load held a file back; and whether the load in progress + # holds one back. Only touched while holding @reload_mutex. + @applied_had_content = {} + @empty_files = {} + @read_again_pending = false + @holding_this_load = false + # The delay of the last retry that was armed. Only tests read it. Guarded by @mutex. + @last_armed_retry_delay = nil + end + + # + # Synchronously loads the files and applies the result, or reports the failure. Use it for + # the initial load. A failure here schedules the same automatic retry as a failed + # triggered reload, and a load that holds back a file it finds empty arms the retry for when + # the file may be read as it is. + # + # @return [Boolean] true if the load succeeded, false if it failed or held back an empty file + # + def reload_now + ensure_started + retry_in = reload(retrying: false) + request_retry(retry_in) unless retry_in.nil? + retry_in.nil? + end + + # + # The delay in seconds of the last retry that was armed, from the time it was armed. Only + # tests use it. + # + # @return [Numeric, nil] + # + def last_armed_retry_delay + @mutex.synchronize { @last_armed_retry_delay } + end + + # + # Signals that the files may have changed and a reload should happen after the debounce + # delay. It never blocks. Signals that arrive while a reload is already pending are + # coalesced. + # + def trigger + ensure_started + @mutex.synchronize do + @trigger_pending = true + @cond.signal + end + end + + # + # Stops the reloader. It does not wait for a reload that is already in progress. A reload + # wedged in a blocking file read must not be able to wedge shutdown. Such a reload can + # still deliver its result through `apply` or `on_error` shortly after this method + # returns, and consumers tolerate that. A reload that has not yet reached its callbacks + # when this method is called does not invoke them. + # + def stop + @mutex.synchronize do + @stopped = true + @cond.broadcast + end + end + + private def ensure_started + @mutex.synchronize do + return if @stopped || !@worker.nil? + + @worker = Thread.new { run } + @worker.name = "LD/FileDataReloader" + end + end + + private def request_retry(delay) + return unless @retry_delay > 0 + + @mutex.synchronize do + @requested_retry_delay = [@requested_retry_delay, delay].compact.min + @cond.signal + end + end + + private def run + loop do + action = next_action + break if action == :stop + + # An unexpected error must not end the worker: no reload would happen after it. + begin + @logger.info { "[LDClient] Reloading flag data after detecting a change" } if action == :reload + retry_in = reload(retrying: action == :retry) + @mutex.synchronize do + # A pending retry is superseded by this reload. The reload either succeeded, or it + # failed or held back a file and arms a fresh retry here. + if retry_in.nil? || @retry_delay <= 0 + @retry_deadline = nil + else + @retry_deadline = monotonic_now + retry_in + @last_armed_retry_delay = retry_in + end + end + rescue => e + log_unexpected_error(e) + end + end + end + + private def log_unexpected_error(error) + Util.log_exception(@logger, "Unexpected error in file data reloader", error) + end + + # + # Waits until a reload is due. Returns :reload for a change-triggered reload, :retry for + # an automatic retry, or :stop. + # + private def next_action + @mutex.synchronize do + loop do + return :stop if @stopped + + now = monotonic_now + if @trigger_pending + @trigger_pending = false + return :reload if @debounce_delay <= 0 + + @debounce_deadline = now + @debounce_delay + end + unless @requested_retry_delay.nil? + # A synchronous reload_now failed or held back a file. Arm the retry without + # reloading again at once. An already armed retry that comes sooner is kept. + deadline = now + @requested_retry_delay + @requested_retry_delay = nil + @retry_deadline = [@retry_deadline, deadline].compact.min + @last_armed_retry_delay = @retry_deadline - now + end + if !@debounce_deadline.nil? && now >= @debounce_deadline + @debounce_deadline = nil + return :reload + end + if !@retry_deadline.nil? && now >= @retry_deadline + @retry_deadline = nil + return :retry + end + + deadlines = [@debounce_deadline, @retry_deadline].compact + timeout = deadlines.empty? ? nil : [deadlines.min - now, 0].max + @cond.wait(@mutex, timeout) + end + end + end + + # + # Performs one full load of all configured files. Returns nil if it succeeded, or the + # seconds after which to retry if it failed or held back a file. A skipped no-op application + # counts as success. The whole set is re-read on every reload: entries are combined across + # files in order, so a change to one file can alter which file wins for a key. + # + private def reload(retrying: false) + @reload_mutex.synchronize do + # A trigger already queued when stop was called can still reach here. + return nil if stopped? + + if retrying + if @read_again_pending + @logger.debug { "[LDClient] Reading flag data again after a file was found empty" } + else + @logger.debug { "[LDClient] Retrying flag data load after earlier failure" } + end + end + @read_again_pending = false + @holding_this_load = false + # The files that the load read and did not hold back, including those whose read or parse + # failed. A load reads every file unless an error it does not expect stops it. + reached = [] + retry_in = + begin + load_files(reached) + rescue => e + # A failure that the load does not expect is retried like any other. + log_unexpected_error(e) + stopped? ? nil : @retry_delay + end + unless @holding_this_load + # A load that holds nothing back, whether it applies, is skipped or fails, starts the + # bound over for every file that it read: such a file is no longer held back, so its + # next rewrite is a new one. + reached.each do |path| + empty = @empty_files[path] + empty.first_held = nil unless empty.nil? + end + end + retry_in + end + end + + # The body of {#reload}, while holding @reload_mutex. + private def load_files(reached) + documents = [] + files = [] + had_content = {} + held = [] + # The first read or parse failure. The other files are still read, so that each of them is + # classified, and the load fails with it after them. + failure = nil + digest = Digest::SHA256.new + @paths.each do |path| + begin + # One read feeds both the digest and the parse, so the skip-unchanged digest can never + # disagree with the content that was applied. + content = FileData.read_file(path) + rescue ReadError => e + if e.missing + not_empty(path, reached) + else + reached << path + end + if e.missing && @skip_missing_paths + @logger.debug { "[LDClient] File #{path} does not exist; it contributes no data" } + had_content[path] = false + append_to_digest(digest, path, nil) + files << FileSummary.new(path, false, 0, 0) + next + end + failure ||= e + next + end + # A file with no content, or YAML that the parser reads as nothing, such as only a + # comment, has no entries. A read can see that much of a file that is being written. + # Content with no JSON document is only parsed when the file is not held back, and then + # fails to parse. + document = nil + has_content = !FileData.blank?(content) + if has_content + begin + document = FileData.parse_file(path, content) + rescue ReadError => e + # Content that fails to parse is content: the file's empty state is over. + not_empty(path, reached) + failure ||= e + next + end + has_content = !document.equal?(Document::NONE) + end + if has_content + not_empty(path, reached) + elsif hold_back?(path, @clock.call) + # The clock is read after this file's own read, so a slow read of another file does + # not shorten its delay. + # The other files are still read, so that files emptied together start their delays + # in the same load. + @holding_this_load = true + held << path + next + else + reached << path + end + had_content[path] = has_content + + append_to_digest(digest, path, content) + if document.nil? + begin + document = FileData.parse_file(path, content) + rescue ReadError => e + failure ||= e + next + end + end + documents << document + files << FileSummary.new(path, true, 0, 0) + end + return record_failure(failure) unless failure.nil? + return hold_back(held, @clock.call) unless held.empty? + + begin + merged = FileData.merge(documents, + duplicate_keys_handling: @duplicate_keys_handling, + logger: @logger) + rescue => e + return record_failure(e) + end + + # Documents are the present files in order. Copy their counts onto the file summaries. + document_index = 0 + files.each do |file| + next unless file.present + + summary = merged.documents[document_index] + file.flags = summary.flags + file.segments = summary.segments + document_index += 1 + end + merged.files = files + + # stop may have been called while the files were being read. Deliver nothing then. + return nil if stopped? + + # A success right after a failure must apply even when the content is unchanged since the + # last success. The consumer heard about the failure through on_error and may have moved + # to an interrupted state. Only apply tells it that things are good again. + recovering = !@last_error_message.nil? + hexdigest = digest.hexdigest + if @skip_unchanged && !recovering && hexdigest == @last_good_digest + remember_applied(had_content) + return nil + end + + # Nothing is remembered until the consumer has accepted the result. A result that apply + # rejects must not become the baseline that skip-unchanged compares against, and must not + # count as a recovery. + begin + @apply.call(merged) + rescue => e + return record_failure(e) + end + @last_error_message = nil + @last_good_digest = hexdigest + remember_applied(had_content) + nil + end + + # Adds a file to the skip-unchanged digest: its path, whether it is present, and its content. + # The same content in another file can parse differently, since the format comes from the + # name, and does not come from the same file. + private def append_to_digest(digest, path, content) + digest << path.b << "\0" + if content.nil? + digest << "-\0" + else + digest << "+" << content.bytesize.to_s << "\0" << content.b + end + end + + # Records that a load found the file with content, even content that fails to parse, or + # missing: its empty state is over, so its next emptying starts a new delay even where a + # coarse modification time makes it look the same. Its bound goes on, so that files + # rewritten in turn stay bounded. + private def not_empty(path, reached) + reached << path + empty = @empty_files[path] + empty.since = nil unless empty.nil? + end + + # Records what each file contributed to the data now in effect. The holds of the files start + # over. + private def remember_applied(had_content) + @applied_had_content = had_content + @empty_files.clear + end + + # + # Decides whether a load that found a file with no content holds it back, because it may be + # in the middle of an in-place rewrite, and records since when it has been so. `now` is read + # after the file's own read. + # + private def hold_back?(path, now) + return false if @retry_delay <= 0 || !@applied_had_content[path] + + state = Poller.observe(path) + empty = (@empty_files[path] ||= EmptyFile.new(nil, nil, nil)) + if empty.since.nil? || empty.state != state + # Found empty for the first time since it had content, or written since a load last + # found it so: a new rewrite. + empty.since = now + empty.state = state + end + # After the delay the file is parsed. The entry stays, so later loads keep parsing it until + # it has content. Its bound starts over only at a load that holds nothing back. + return false if now - empty.since >= @retry_delay + return false if !empty.first_held.nil? && now - empty.first_held >= MAX_HOLD_RETRY_DELAYS * @retry_delay + + empty.first_held ||= now + true + end + + # + # Holds back a load that found files with no content. The current data stays and no failure + # is reported. Returns when to retry: when the delay or the bound of the first of the files + # is over, plus a tenth of the retry delay, so that a retry that comes a little early still + # finds it over. + # + private def hold_back(paths, now) + # stop may have been called while the files were being read. Arm no retry. + return nil if stopped? + + @read_again_pending = true + retry_at = nil + paths.each do |path| + @logger.debug do + "[LDClient] File #{path} is empty, possibly while it is being rewritten; reading it again before applying it" + end + empty = @empty_files[path] + over = [empty.since + @retry_delay, empty.first_held + MAX_HOLD_RETRY_DELAYS * @retry_delay].min + retry_at = over if retry_at.nil? || over < retry_at + end + [retry_at - now, 0].max + @retry_delay / 10.0 + end + + private def record_failure(error) + # stop may have been called while the files were being read. Deliver nothing then, and + # arm no retry. + return nil if stopped? + + # With automatic retries, a persistent failure would repeat the same log entry and the + # same callback on every attempt. Repeats of an identical failure are logged at debug + # level and do not re-invoke on_error. + message = error.message + if message == @last_error_message + @logger.debug { "[LDClient] Unable to load flags: #{message}" } + return @retry_delay + end + + @last_error_message = message + @logger.error { "[LDClient] Unable to load flags: #{message}" } + @on_error&.call(error) + @retry_delay + end + + private def stopped? + @mutex.synchronize { @stopped } + end + + private def monotonic_now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + end + end + end +end diff --git a/lib/ldclient-rb/impl/file_data/watcher.rb b/lib/ldclient-rb/impl/file_data/watcher.rb new file mode 100644 index 00000000..73c0add6 --- /dev/null +++ b/lib/ldclient-rb/impl/file_data/watcher.rb @@ -0,0 +1,349 @@ +# frozen_string_literal: true + +require "ldclient-rb/impl/repeating_task" +require "ldclient-rb/impl/util" + +require "concurrent/atomics" +require "set" + +module LaunchDarkly + module Impl + module FileData + # + # Detects changes to a set of files through file system change notifications. The + # notification mechanism comes from the optional `listen` gem, which the SDK does not depend + # on. Check {Watcher.available?} before constructing a watcher. + # + # On Linux the watcher uses `rb-inotify`, which `listen` depends on, to watch the directory of + # each file without descending into subdirectories. Elsewhere it uses `listen` itself, which + # scans the whole directory tree under each watched directory. + # + # The watcher observes the directory of each file, so a configured file that does not exist + # yet is picked up when it appears. Each directory is watched on its own: one that does not + # exist, or that is deleted while it is watched, is logged and retried once per second until + # it exists, while the other directories stay watched. When a retry sets up a watch, the + # callback runs once, so that a change made while there was no watch is not missed. + # + # @private + # + class Watcher + # Seconds between attempts to set up the watches after a failure. + RETRY_INTERVAL = 1.0 + + INOTIFY_EVENTS = [:create, :modify, :close_write, :attrib, :delete, :moved_to, :moved_from, + :delete_self, :move_self].freeze + private_constant :INOTIFY_EVENTS + + # The inotify event flags that mean the watched directory itself is gone. + INOTIFY_DIRECTORY_LOST = [:delete_self, :move_self].freeze + private_constant :INOTIFY_DIRECTORY_LOST + + # The watch on one real directory: the configured directories that resolve to it, the + # names of the configured files in it, and a callable that removes the watch. + WatchedDirectory = Struct.new(:directories, :names, :close) + private_constant :WatchedDirectory + + # + # Returns true if a change notification mechanism can be loaded. + # + # @return [Boolean] + # + def self.available? + inotify_available? || listen_available? + end + + # + # Returns true if `rb-inotify` can be loaded. It is a dependency of `listen` on Linux and + # does not load on other platforms. + # + # @return [Boolean] + # + def self.inotify_available? + @inotify_available = load_library("rb-inotify") if @inotify_available.nil? + @inotify_available + end + + # + # Returns true if the `listen` gem can be loaded. + # + # @return [Boolean] + # + def self.listen_available? + @listen_available = load_library("listen") if @listen_available.nil? + @listen_available + end + + private_class_method def self.load_library(name) + require name + true + rescue LoadError, StandardError + false + end + + # + # Creates and starts a watcher. + # + # @param paths [Array] absolute paths of the files to watch + # @param on_change [#call] invoked with no arguments when one of the files changes + # @param logger [Logger] + # + def initialize(paths, on_change, logger) + @paths = paths + @on_change = on_change + @logger = logger + @stopped = Concurrent::AtomicBoolean.new(false) + @lock = Mutex.new + @retry_task = nil + @last_error_message = nil + # The directories of the configured paths, in configuration order. A directory is in + # @missing until it is watched, and then in @watched under its real path. The lock + # guards both, @retry_task, and @inotify. + @directories = paths.map { |p| File.dirname(p) }.uniq + @missing = Set.new(@directories) + @watched = {} + @inotify = nil + + try_start + schedule_retry unless @lock.synchronize { @missing.empty? } + end + + # + # Stops the watcher. No callback runs after this method returns, apart from one that is + # already in progress. + # + def stop + return unless @stopped.make_true + + closers, retry_task, inotify = @lock.synchronize do + state = [@watched.values.map(&:close), @retry_task, @inotify] + @watched.clear + @retry_task = nil + @inotify = nil + state + end + retry_task&.stop + closers.each(&:call) + inotify&.stop + end + + # + # Starts the task that attempts to watch the missing directories once per second, unless + # it already runs or the watcher is stopped. + # + private def schedule_retry + @lock.synchronize do + return if @stopped.value || !@retry_task.nil? + + @retry_task = RepeatingTask.new(RETRY_INTERVAL, RETRY_INTERVAL, method(:retry_start), @logger, + "LD/FileDataWatcherRetry") + @retry_task.start + end + end + + private def retry_start + return if @stopped.value + + added = try_start + # The task ends when nothing is missing. A directory lost meanwhile is back in @missing + # before its loss report calls schedule_retry, so either the task is kept here or that + # call starts a new one. + retry_task = @lock.synchronize do + next nil unless @missing.empty? + + task = @retry_task + @retry_task = nil + task + end + # This runs on the retry task's own thread, which RepeatingTask#stop allows. + retry_task&.stop + @on_change.call if added && !@stopped.value + end + + # + # Handles the loss of a watched directory. The notification mechanism reports it on its + # own thread. The watch ends with the directory, so it is removed and set up again through + # the same retry as at start, once the directory exists. The other directories keep their + # watches. + # + private def directory_lost(real_directory) + entry = @lock.synchronize do + return if @stopped.value + + e = @watched.delete(real_directory) + return if e.nil? + + @missing.merge(e.directories) + e + end + @logger.warn { "[LDClient] Directory #{real_directory} no longer exists; its data files are watched again when it exists" } + entry.close.call + schedule_retry + end + + # + # Watches each missing directory that exists now. Returns true if at least one watch was + # added. The directories that are still missing, and any watch that could not be set up, + # are logged together, so that an unchanged situation repeats at debug level. + # + private def try_start + added = false + missing = [] + problems = [] + @lock.synchronize { @directories.select { |d| @missing.include?(d) } }.each do |directory| + unless File.directory?(directory) + missing << directory + next + end + begin + added = true if watch_directory(directory) + rescue => e + problems << e.message + end + end + problems.unshift("directory does not exist: #{missing.join(', ')}") unless missing.empty? + if problems.empty? + @last_error_message = nil + else + log_setup_failure(problems.join("; ")) + end + added + end + + # + # Watches the real directory of a configured directory, and records the names of the + # configured files in it. Two configured directories can resolve to the same real + # directory, which then has one watch for the names in both. Returns false if the watcher + # is stopped. + # + private def watch_directory(directory) + real_directory = File.realpath(directory) + names = @paths.select { |p| File.dirname(p) == directory }.map { |p| File.basename(p) } + @lock.synchronize do + return false if @stopped.value + + entry = @watched[real_directory] + if entry.nil? + entry = WatchedDirectory.new(Set.new, Set.new, add_watch(real_directory)) + @watched[real_directory] = entry + end + entry.directories << directory + entry.names.merge(names) + @missing.delete(directory) + end + true + end + + # + # Adds the notification mechanism's watch on a real directory and returns a callable that + # removes it. Called with the lock held, so that stop sees either no watch or a recorded + # one. + # + private def add_watch(real_directory) + Watcher.inotify_available? ? add_inotify_watch(real_directory) : start_listen(real_directory) + end + + # + # Adds a watch on the directory itself, without descending into subdirectories, to the one + # inotify notifier, which is created with the first watch. + # + private def add_inotify_watch(real_directory) + @inotify ||= InotifyListener.new(INotify::Notifier.new, @logger) + watch = @inotify.watch(real_directory) { |event| inotify_event(real_directory, event) } + lambda do + begin + watch.close + rescue SystemCallError + # The kernel already removed the watch along with the directory. + end + end + end + + private def inotify_event(real_directory, event) + return if @stopped.value + + if (event.flags & INOTIFY_DIRECTORY_LOST).empty? + @on_change.call if watched_name?(real_directory, event.name) + else + directory_lost(real_directory) + end + end + + # + # Watches a real directory with the `listen` gem, which reports paths under the real + # directory, so a reported path is matched by its directory and name. + # + private def start_listen(real_directory) + listener = Listen.to(real_directory) do |modified, added, removed| + next if @stopped.value + + changed = (modified + added + removed).any? do |p| + File.dirname(p) == real_directory && watched_name?(real_directory, File.basename(p)) + end + @on_change.call if changed && !@stopped.value + end + listener.start + -> { listener.stop } + end + + # + # Returns true if the name is one of the configured files in the watched real directory. + # + private def watched_name?(real_directory, name) + @lock.synchronize do + entry = @watched[real_directory] + !entry.nil? && entry.names.include?(name) + end + end + + private def log_setup_failure(message) + if message == @last_error_message + @logger.debug { "[LDClient] Unable to watch data files: #{message}" } + else + @last_error_message = message + @logger.error { "[LDClient] Unable to watch data files: #{message}" } + end + end + + # + # Runs an inotify notifier on its own thread, takes watches for it over time, and stops it + # on request. + # + class InotifyListener + def initialize(notifier, logger) + @notifier = notifier + @thread = Thread.new do + begin + notifier.run + rescue IOError, SystemCallError + # The notifier was closed by stop. + rescue => e + Util.log_exception(logger, "Unexpected error in file data watcher", e) + end + end + @thread.name = "LD/FileDataWatcher" + end + + # + # Adds a watch on a directory and returns the notifier's watch object, whose `close` + # removes the watch again. + # + def watch(directory, &callback) + @notifier.watch(directory, *INOTIFY_EVENTS, &callback) + end + + # + # Stops the notifier and waits briefly for its thread. Closing the notifier ends the + # blocking read that the thread is in. A callback can call this on the notifier's own + # thread, which then ends when the callback returns. + # + def stop + @notifier.stop + @notifier.close + @thread.join(2) unless Thread.current == @thread + end + end + private_constant :InotifyListener + end + end + end +end diff --git a/spec/impl/file_data/document_spec.rb b/spec/impl/file_data/document_spec.rb new file mode 100644 index 00000000..544e88b2 --- /dev/null +++ b/spec/impl/file_data/document_spec.rb @@ -0,0 +1,308 @@ +# frozen_string_literal: true + +require "spec_helper" +require "tmpdir" +require "ldclient-rb/impl/file_data" + +module LaunchDarkly + module Impl + module FileData + describe Document do + describe "parse" do + it "parses a JSON document with every section and symbolizes nested keys" do + document = Document.parse(<<~JSON, format: :json) + { + "flags": { "flag1": { "key": "flag1", "on": true, "rules": [ { "clauses": [ { "op": "in" } ] } ] } }, + "flagValues": { "flag2": "value2" }, + "segments": { "seg1": { "key": "seg1", "included": ["user1"] } } + } + JSON + + expect(document.flags.keys).to eq([:flag1]) + expect(document.flags[:flag1][:rules][0][:clauses][0][:op]).to eq("in") + expect(document.flag_values).to eq({ flag2: "value2" }) + expect(document.segments[:seg1][:included]).to eq(["user1"]) + end + + it "parses a YAML document" do + document = Document.parse(<<~YAML, format: :yaml) + --- + flags: + flag1: + key: flag1 + "on": true + flagValues: + flag2: value2 + segments: + seg1: + key: seg1 + YAML + + expect(document.flags[:flag1][:on]).to be true + expect(document.flag_values).to eq({ flag2: "value2" }) + expect(document.segments.keys).to eq([:seg1]) + end + + it "treats empty YAML content as a document with no entries" do + ["", " \n\r\n", "# a comment\n", "---\n", "~\n"].each do |content| + document = Document.parse(content, format: :yaml) + + expect(document.flags).to eq({}) + expect(document.flag_values).to eq({}) + expect(document.segments).to eq({}) + end + end + + it "returns NONE for YAML that is only a null, and a document for {} or a member that is null" do + ["", "# a comment\n", "---\n", "~\n"].each do |content| + expect(Document.parse(content, format: :yaml)).to be(Document::NONE) + end + ["{}\n", "flagValues:\n"].each do |content| + document = Document.parse(content, format: :yaml) + + expect(document).not_to be(Document::NONE) + expect(document.flag_values).to eq({}) + end + expect(Document.parse("{}", format: :json)).not_to be(Document::NONE) + end + + it "treats YAML with only white space, including a tab, vertical tab or form feed, as no entries" do + expect(Document.parse("\t\n", format: :yaml)).to be(Document::NONE) + expect(Document.parse(" \v\f\t\r\n", format: :yaml)).to be(Document::NONE) + end + + it "rejects content that is not valid UTF-8, in both formats" do + latin1 = (+"{\"flagValues\": {\"flag1\": \"caf\xE9\"}}").force_encoding(Encoding::UTF_8) + [:json, :yaml].each do |format| + expect { Document.parse(latin1, format: format) }.to raise_error(ArgumentError, "the file is not valid UTF-8") + end + end + + it "rejects empty JSON content, which is not a JSON document" do + ["", " \n\r\n\t"].each do |content| + expect { Document.parse(content, format: :json) } + .to raise_error(ArgumentError, "the file has no content, which is not a JSON document") + end + end + + it "treats a document with no known sections as a document with no entries" do + [[:json, "{}"], [:yaml, "{}"], [:yaml, "other: 1\n"]].each do |format, content| + document = Document.parse(content, format: format) + + expect(document.flags).to eq({}) + expect(document.flag_values).to eq({}) + expect(document.segments).to eq({}) + end + end + + it "rejects a document that is not an object" do + [:json, :yaml].each do |format| + expect { Document.parse("\"hello\"", format: format) }.to raise_error(ArgumentError, /must be an object/) + expect { Document.parse("[1, 2]", format: format) }.to raise_error(ArgumentError, /must be an object/) + end + expect { Document.parse("null", format: :json) }.to raise_error(ArgumentError, /must be an object/) + expect { Document.parse("false", format: :yaml) }.to raise_error(ArgumentError, /must be an object/) + end + + it "rejects a section that is not an object" do + expect { Document.parse('{"flags": []}', format: :json) }.to raise_error(ArgumentError, /"flags" must be an object/) + expect { Document.parse('{"flagValues": 3}', format: :json) }.to raise_error(ArgumentError, /"flagValues" must be an object/) + expect { Document.parse('{"segments": "x"}', format: :json) }.to raise_error(ArgumentError, /"segments" must be an object/) + end + + it "symbolizes keys that YAML parsed as numbers" do + document = Document.parse("flagValues:\n 123: true\n", format: :yaml) + + expect(document.flag_values).to eq({ "123": true }) + end + + it "reads JSON numbers as JSON does" do + document = Document.parse('{"flagValues": {"flag1": 1e3, "flag2": 12}}', format: :json) + + expect(document.flag_values).to eq({ flag1: 1000.0, flag2: 12 }) + end + + it "raises a syntax error for content that cannot be parsed" do + expect { Document.parse('{"flagValues"', format: :json) }.to raise_error(JSON::ParserError) + expect { Document.parse('{"flagValues": {}} extra', format: :json) }.to raise_error(JSON::ParserError) + expect { Document.parse("flagValues: [", format: :yaml) }.to raise_error(Psych::SyntaxError) + end + + it "rejects a YAML document as JSON" do + expect { Document.parse("flagValues:\n flag1: true\n", format: :json) }.to raise_error(JSON::ParserError) + end + + it "rejects an unknown format" do + expect { Document.parse("{}", format: :toml) }.to raise_error(ArgumentError, /unknown file format/) + end + end + + describe "read" do + around do |example| + Dir.mktmpdir do |dir| + @dir = dir + example.run + end + end + + it "reads and parses a file" do + path = File.join(@dir, "flags.json") + File.write(path, '{"flagValues": {"flag1": 1}}') + + document = Document.read(path) + + expect(document.flag_values).to eq({ flag1: 1 }) + end + + it "reads a file as YAML when its name ends in .yaml or .yml, in any case" do + ["flags.yaml", "flags.yml", "flags.YAML", "flags.Yml"].each do |name| + path = File.join(@dir, name) + File.write(path, "flagValues:\n flag1: true\n") + + expect(Document.read(path).flag_values).to eq({ flag1: true }) + end + end + + it "reads any other file as JSON" do + ["flags.json", "flags", "flags.txt", "flags.yaml.json"].each do |name| + path = File.join(@dir, name) + File.write(path, "flagValues:\n flag1: true\n") + + expect { Document.read(path) }.to raise_error(ReadError, /error parsing file/) + end + end + + it "reports an empty file as a read error when it is JSON, and as no entries when it is YAML" do + ["flags.json", "flags"].each do |name| + path = File.join(@dir, name) + File.write(path, "") + + expect { Document.read(path) }.to raise_error(ReadError, /the file has no content, which is not a JSON document/) + end + + path = File.join(@dir, "flags.yaml") + File.write(path, "") + expect(Document.read(path).flag_values).to eq({}) + end + + it "reads a file that starts with a byte order mark" do + path = File.join(@dir, "flags.json") + File.binwrite(path, "\xEF\xBB\xBF{\"flagValues\": {\"flag1\": \"caf\xC3\xA9\"}}") + + expect(Document.read(path).flag_values).to eq({ flag1: "caf\u00e9" }) + end + + it "reports a file encoded in UTF-16 or UTF-32 as a read error" do + path = File.join(@dir, "flags.json") + content = '{"flagValues": {"flag1": true}}' + ["\xFF\xFE".b + content.encode("UTF-16LE").b, "\xFE\xFF".b + content.encode("UTF-16BE").b, + "\xFF\xFE\x00\x00".b + content.encode("UTF-32LE").b].each do |bytes| + File.binwrite(path, bytes) + + expect { Document.read(path) }.to raise_error(ReadError, /encoded in UTF-16 or UTF-32; data files must be UTF-8/) + end + end + + it "reads a file as UTF-8 whatever the default external encoding" do + path = File.join(@dir, "flags.yaml") + File.binwrite(path, "flagValues:\n flag1: caf\xC3\xA9\n") + previous = Encoding.default_external + begin + Encoding.default_external = Encoding::US_ASCII + expect(Document.read(path).flag_values).to eq({ flag1: "caf\u00e9" }) + ensure + Encoding.default_external = previous + end + end + + it "reads a file that holds only a byte order mark as empty" do + path = File.join(@dir, "flags.json") + File.binwrite(path, "\xEF\xBB\xBF") + + expect(FileData.read_file(path)).to eq("") + end + + it "reports a missing file as a read error that is marked missing" do + path = File.join(@dir, "no-such-file.json") + + expect { Document.read(path) }.to raise_error(ReadError) do |error| + expect(error.missing).to be true + expect(error.path).to eq(path) + expect(error.message).to include(path) + end + end + + it "reports a path that cannot be read as a read error that is not marked missing" do + expect { Document.read(@dir) }.to raise_error(ReadError) do |error| + expect(error.missing).to be false + expect(error.path).to eq(@dir) + end + end + + it "reports unparseable content as a read error that is not marked missing" do + path = File.join(@dir, "bad.json") + File.write(path, '{"flagValues"') + + expect { Document.read(path) }.to raise_error(ReadError) do |error| + expect(error.missing).to be false + expect(error.message).to include("error parsing file") + expect(error.message).to include(path) + end + end + + it "reports a document that is not an object as a read error" do + path = File.join(@dir, "scalar.yaml") + File.write(path, "just a string\n") + + expect { Document.read(path) }.to raise_error(ReadError, /must be an object/) + end + end + end + + describe "format_of" do + it "chooses YAML for a name that ends in .yaml or .yml, in any case, and JSON otherwise" do + expect(FileData.format_of("/a/flags.yaml")).to eq(:yaml) + expect(FileData.format_of("/a/.yaml")).to eq(:yaml) + expect(FileData.format_of("/a/.yml")).to eq(:yaml) + expect(FileData.format_of("/a/flags.YML")).to eq(:yaml) + expect(FileData.format_of("/a/flags.json")).to eq(:json) + expect(FileData.format_of("/a.yaml/flags")).to eq(:json) + expect(FileData.format_of("/a/flags.yaml.bak")).to eq(:json) + end + end + + describe "blank?" do + it "is true for content that is empty or only white space" do + expect(FileData.blank?("")).to be true + expect(FileData.blank?(" \t\r\n\f\v")).to be true + expect(FileData.blank?(" {}")).to be false + expect(FileData.blank?("# comment")).to be false + expect(FileData.blank?("\xFF".b)).to be false + end + end + + describe "make_flag_with_value" do + it "builds a flag that is on and serves the value as its fallthrough" do + flag = FileData.make_flag_with_value("flag1", "value1") + + expect(flag).to eq({ + key: "flag1", + on: true, + version: 1, + fallthrough: { variation: 0 }, + variations: ["value1"], + }) + end + end + + describe "absolute_paths" do + it "converts relative paths to absolute paths and accepts a single string" do + absolute = File.absolute_path("/x/y.json") + + expect(FileData.absolute_paths("a/b.json")).to eq([File.absolute_path("a/b.json")]) + expect(FileData.absolute_paths([absolute, "z.json"])).to eq([absolute, File.absolute_path("z.json")]) + end + end + end + end +end diff --git a/spec/impl/file_data/merge_spec.rb b/spec/impl/file_data/merge_spec.rb new file mode 100644 index 00000000..9fd23cb5 --- /dev/null +++ b/spec/impl/file_data/merge_spec.rb @@ -0,0 +1,132 @@ +# frozen_string_literal: true + +require "spec_helper" +require "capturing_logger" +require "ldclient-rb/impl/file_data" + +module LaunchDarkly + module Impl + module FileData + describe "merge" do + def document(flags: {}, flag_values: {}, segments: {}) + Document.new(flags: flags, flag_values: flag_values, segments: segments) + end + + def full_flag(key, version: nil) + data = { key: key, on: true, variations: ["a", "b"], fallthrough: { variation: 1 } } + data[:version] = version unless version.nil? + data + end + + it "combines flags, flag values, and segments from documents in order" do + doc1 = document(flags: { flag1: full_flag("flag1") }, segments: { seg1: { key: "seg1", included: ["u"] } }) + doc2 = document(flag_values: { flag2: "value2" }) + + result = FileData.merge([doc1, doc2]) + + expect(result.flags.keys).to eq([:flag1, :flag2]) + expect(result.flags[:flag1]).to be_a(Model::FeatureFlag) + expect(result.flags[:flag2]).to be_a(Model::FeatureFlag) + expect(result.segments.keys).to eq([:seg1]) + expect(result.segments[:seg1]).to be_a(Model::Segment) + expect(result.documents).to eq([DocumentSummary.new(1, 1), DocumentSummary.new(1, 0)]) + expect(result.empty?).to be false + end + + it "expands a flag value into a flag that serves the value by fallthrough for every context" do + result = FileData.merge([document(flag_values: { flag2: "value2" })]) + + flag = result.flags[:flag2] + expect(flag.key).to eq("flag2") + expect(flag.on).to be true + expect(flag.version).to eq(1) + expect(flag.variations).to eq(["value2"]) + expect(flag.off_variation).to be_nil + expect(flag.fallthrough.variation).to eq(0) + end + + it "reports an empty result for no documents" do + expect(FileData.merge([]).empty?).to be true + end + + it "fails when the same flag key appears in two documents" do + docs = [document(flags: { flag1: full_flag("flag1") }), document(flags: { flag1: full_flag("flag1") })] + + expect { FileData.merge(docs) }.to raise_error(MergeError, /flag key "flag1" was used more than once/) + end + + it "fails when the same segment key appears in two documents" do + docs = [document(segments: { seg1: { key: "seg1" } }), document(segments: { seg1: { key: "seg1" } })] + + expect { FileData.merge(docs) }.to raise_error(MergeError, /segment key "seg1" was used more than once/) + end + + it "fails when a key appears in both flags and flagValues of one document" do + docs = [document(flags: { flag1: full_flag("flag1") }, flag_values: { flag1: "x" })] + + expect { FileData.merge(docs) }.to raise_error(MergeError, /flag key "flag1"/) + end + + it "keeps the first document's entry and drops later duplicates with ignore handling" do + doc1 = document(flag_values: { flag1: "first" }) + doc2 = document(flag_values: { flag1: "second", flag2: "other" }) + + result = FileData.merge([doc1, doc2], duplicate_keys_handling: DuplicateKeysHandling::IGNORE) + + expect(result.flags[:flag1].variations).to eq(["first"]) + expect(result.flags[:flag2].variations).to eq(["other"]) + expect(result.documents).to eq([DocumentSummary.new(1, 0), DocumentSummary.new(1, 0)]) + end + + it "defaults a missing version to 1 and keeps a given version" do + doc = document( + flags: { flag1: full_flag("flag1"), flag2: full_flag("flag2", version: 5) }, + segments: { seg1: { key: "seg1" }, seg2: { key: "seg2", version: 9 } } + ) + + result = FileData.merge([doc]) + + expect(result.flags[:flag1].version).to eq(1) + expect(result.flags[:flag2].version).to eq(5) + expect(result.segments[:seg1].version).to eq(1) + expect(result.segments[:seg2].version).to eq(9) + end + + it "fills a missing key member from the key the entry appears under" do + doc = document(flags: { flag1: { on: false, variations: [1] } }, segments: { seg1: {} }) + + result = FileData.merge([doc]) + + expect(result.flags[:flag1].key).to eq("flag1") + expect(result.segments[:seg1].key).to eq("seg1") + end + + it "does not modify the document's own hashes" do + data = { on: false, variations: [1] } + FileData.merge([document(flags: { flag1: data })]) + + expect(data).to eq({ on: false, variations: [1] }) + end + + it "fails when an entry is not an object" do + expect { FileData.merge([document(flags: { flag1: 5 })]) }.to raise_error(MergeError, /flag "flag1" is not an object/) + expect { FileData.merge([document(segments: { seg1: [] })]) }.to raise_error(MergeError, /segment "seg1" is not an object/) + end + + it "fails when an entry cannot be deserialized into the data model" do + expect { FileData.merge([document(flags: { flag1: { rules: 5 } })]) }.to raise_error(MergeError, /flag "flag1"/) + expect { FileData.merge([document(segments: { seg1: { rules: 5 } })]) }.to raise_error(MergeError, /segment "seg1"/) + end + + it "passes the logger to model validation" do + logger = CapturingLogger.new + bad = { key: "flag1", on: true, variations: ["a"], fallthrough: { variation: 5 } } + + FileData.merge([document(flags: { flag1: bad })], logger: logger) + + expect(logger.output).to include("Data inconsistency in feature flag \"flag1\"") + end + end + end + end +end diff --git a/spec/impl/file_data/poller_spec.rb b/spec/impl/file_data/poller_spec.rb new file mode 100644 index 00000000..88e820f2 --- /dev/null +++ b/spec/impl/file_data/poller_spec.rb @@ -0,0 +1,119 @@ +# frozen_string_literal: true + +require "spec_helper" +require "tmpdir" +require "ldclient-rb/impl/file_data" + +module LaunchDarkly + module Impl + module FileData + describe Poller do + let(:interval) { 0.05 } + + around do |example| + Dir.mktmpdir do |dir| + @dir = dir + example.run + end + end + + def path(name) + File.join(@dir, name) + end + + def wait_for(timeout = 3) + deadline = Time.now + timeout + until yield + return false if Time.now > deadline + sleep 0.01 + end + true + end + + def with_poller(paths) + calls = Concurrent::AtomicFixnum.new(0) + poller = Poller.new(paths, interval, -> { calls.increment }, $null_log) + begin + yield poller, calls + ensure + poller.stop + end + end + + it "observes the state of the file that a symbolic link leads to" do + skip "symbolic links need extra privileges on Windows" if Gem.win_platform? + File.write(path("target.json"), "{}") + File.symlink(path("target.json"), path("link.json")) + File.utime(Time.at(1_000_000), Time.at(1_000_000), path("target.json")) + + state = Poller.observe(path("link.json")) + + expect(state).to eq(Poller::FileState.new(true, Time.at(1_000_000), 2)) + expect(Poller.observe(path("missing.json")).exists).to be false + end + + it "invokes the callback when a file's modification time changes" do + File.write(path("a.json"), "{}") + with_poller([path("a.json")]) do |_poller, calls| + File.utime(Time.now + 10, Time.now + 10, path("a.json")) + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "invokes the callback when a file's size changes but its modification time does not" do + File.write(path("a.json"), "{}") + mtime = File.mtime(path("a.json")) + with_poller([path("a.json")]) do |_poller, calls| + File.write(path("a.json"), '{"flagValues": {}}') + File.utime(mtime, mtime, path("a.json")) + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "invokes the callback when a file appears" do + with_poller([path("missing.json")]) do |_poller, calls| + sleep interval * 2 + expect(calls.value).to eq(0) + File.write(path("missing.json"), "{}") + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "invokes the callback when a file disappears" do + File.write(path("a.json"), "{}") + with_poller([path("a.json")]) do |_poller, calls| + File.delete(path("a.json")) + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "watches every configured file" do + File.write(path("a.json"), "{}") + File.write(path("b.json"), "{}") + with_poller([path("a.json"), path("b.json")]) do |_poller, calls| + File.utime(Time.now + 10, Time.now + 10, path("b.json")) + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "does not invoke the callback when nothing changed" do + File.write(path("a.json"), "{}") + with_poller([path("a.json")]) do |_poller, calls| + sleep interval * 6 + expect(calls.value).to eq(0) + end + end + + it "does not invoke the callback after it is stopped" do + File.write(path("a.json"), "{}") + with_poller([path("a.json")]) do |poller, calls| + poller.stop + File.utime(Time.now + 10, Time.now + 10, path("a.json")) + sleep interval * 6 + expect(calls.value).to eq(0) + end + end + end + end + end +end diff --git a/spec/impl/file_data/reloader_spec.rb b/spec/impl/file_data/reloader_spec.rb new file mode 100644 index 00000000..194fd1d5 --- /dev/null +++ b/spec/impl/file_data/reloader_spec.rb @@ -0,0 +1,1893 @@ +# frozen_string_literal: true + +require "spec_helper" +require "capturing_logger" +require "in_place_writes" +require "tmpdir" +require "ldclient-rb/impl/file_data" + +module LaunchDarkly + module Impl + module FileData + describe Reloader do + around do |example| + Dir.mktmpdir do |dir| + @dir = dir + example.run + end + end + + def path(name) + File.join(@dir, name) + end + + # Applications rewrite data files in place, which is not atomic. The tests do too, deliberately + # slowly: a file that exists is truncated, and written 30 ms later, so a load can find it + # empty, as it would with an application's writes, instead of only by chance. + def write(name, content) + if File.exist?(path(name)) + InPlaceWrites.rewrite(path(name), content, 0.03) + else + File.write(path(name), content) + end + path(name) + end + + def values_doc(values) + { flagValues: values }.to_json + end + + def wait_for(timeout = 3) + deadline = Time.now + timeout + until yield + return false if Time.now > deadline + sleep 0.01 + end + true + end + + # Collects apply and on_error calls in a thread-safe way. It raises from the first `reject` + # apply calls, as a consumer that cannot use a result does. + class Recorder + attr_reader :applied, :errors + + def initialize(reject: 0) + @lock = Mutex.new + @applied = [] + @errors = [] + @reject = reject + end + + def apply(merged) + @lock.synchronize do + if @reject > 0 + @reject -= 1 + raise ArgumentError, "the result was rejected" + end + @applied << merged + end + end + + def on_error(error) + @lock.synchronize { @errors << error } + end + + def flag_values(index = -1) + @applied[index].flags.transform_values { |flag| flag.variations[0] } + end + end + + def make_reloader(paths, recorder, logger: $null_log, **options) + Reloader.new(paths: paths, logger: logger, apply: recorder.method(:apply), on_error: recorder.method(:on_error), **options) + end + + def with_reloader(paths, logger: $null_log, reject: 0, **options) + recorder = Recorder.new(reject: reject) + reloader = make_reloader(paths, recorder, logger: logger, **options) + begin + yield reloader, recorder + ensure + reloader.stop + end + end + + it "does not start a worker thread until it is used" do + with_reloader([write("a.json", "{}")]) do |reloader, _recorder| + expect(reloader.instance_variable_get(:@worker)).to be_nil + + reloader.trigger + + worker = reloader.instance_variable_get(:@worker) + expect(worker).to be_a(Thread) + expect(worker.name).to eq("LD/FileDataReloader") + end + end + + it "applies the merged files and describes each file on a synchronous reload" do + a = write("a.json", values_doc({ flag1: "a" })) + b = write("b.json", { flagValues: { flag2: "b" }, segments: { seg1: { key: "seg1" } } }.to_json) + + with_reloader([a, b]) do |reloader, recorder| + expect(reloader.reload_now).to be true + + expect(recorder.applied.length).to eq(1) + expect(recorder.flag_values).to eq({ flag1: "a", flag2: "b" }) + expect(recorder.applied[0].segments.keys).to eq([:seg1]) + expected_files = [FileSummary.new(a, true, 1, 0), FileSummary.new(b, true, 1, 1)] + expect(recorder.applied[0].files).to eq(expected_files) + expect(recorder.errors).to be_empty + end + end + + it "fails a reload for a missing file by default" do + with_reloader([path("missing.json")], retry_delay: 0) do |reloader, recorder| + expect(reloader.reload_now).to be false + + expect(recorder.applied).to be_empty + expect(recorder.errors.length).to eq(1) + expect(recorder.errors[0]).to be_a(ReadError) + expect(recorder.errors[0].missing).to be true + end + end + + it "treats a missing file as a file with no content when skipping missing paths" do + a = write("a.json", values_doc({ flag1: "a" })) + missing = path("missing.json") + + with_reloader([a, missing], skip_missing_paths: true) do |reloader, recorder| + expect(reloader.reload_now).to be true + + expect(recorder.flag_values).to eq({ flag1: "a" }) + expected_files = [FileSummary.new(a, true, 1, 0), FileSummary.new(missing, false, 0, 0)] + expect(recorder.applied[0].files).to eq(expected_files) + end + end + + it "applies an empty result when every file is missing and missing paths are skipped" do + with_reloader([path("missing.json")], skip_missing_paths: true) do |reloader, recorder| + expect(reloader.reload_now).to be true + + expect(recorder.applied.length).to eq(1) + expect(recorder.applied[0].empty?).to be true + end + end + + it "keeps the last good result when a file cannot be parsed and reports the failure once" do + a = write("a.json", values_doc({ flag1: "a" })) + logger = CapturingLogger.new + + with_reloader([a], logger: logger, retry_delay: 0) do |reloader, recorder| + reloader.reload_now + write("a.json", '{"flagValues"') + + expect(reloader.reload_now).to be false + expect(reloader.reload_now).to be false + + expect(recorder.applied.length).to eq(1) + expect(recorder.errors.length).to eq(1) + expect(recorder.errors[0]).to be_a(ReadError) + expect(logger.output.scan("ERROR").length).to eq(1) + end + end + + it "reports a different failure again" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], retry_delay: 0) do |reloader, recorder| + write("a.json", '{"flagValues"') + reloader.reload_now + write("a.json", '{"flagValues": []}') + reloader.reload_now + + expect(recorder.errors.length).to eq(2) + end + end + + it "fails a reload for duplicate keys across files and applies the first file's entry with ignore handling" do + a = write("a.json", values_doc({ flag1: "first" })) + b = write("b.json", values_doc({ flag1: "second" })) + + with_reloader([a, b], retry_delay: 0) do |reloader, recorder| + expect(reloader.reload_now).to be false + expect(recorder.errors[0]).to be_a(MergeError) + end + + with_reloader([a, b], duplicate_keys_handling: DuplicateKeysHandling::IGNORE) do |reloader, recorder| + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag1: "first" }) + end + end + + it "retries a failed reload after the retry delay and recovers when the file is fixed" do + a = write("a.json", '{"flagValues"') + + with_reloader([a], retry_delay: 0.1) do |reloader, recorder| + expect(reloader.reload_now).to be false + write("a.json", values_doc({ flag1: "fixed" })) + + # The retry recovers with the fixed content. The write is in place, so a retry can + # find the file empty while it is being written. It reads the file again later + # instead of applying it. + expect(wait_for { recorder.applied.length >= 1 }).to be true + expect(recorder.applied.map { |merged| merged.flags.keys }).to all(eq([:flag1])) + expect(recorder.flag_values).to eq({ flag1: "fixed" }) + end + end + + it "keeps retrying while the failure persists" do + a = write("a.json", '{"flagValues"') + logger = CapturingLogger.new + + with_reloader([a], logger: logger, retry_delay: 0.05) do |reloader, _recorder| + reloader.reload_now + + expect(wait_for { logger.output.scan("Retrying flag data load").length >= 3 }).to be true + end + end + + it "does not retry when the retry delay is zero" do + a = write("a.json", '{"flagValues"') + + with_reloader([a], retry_delay: 0) do |reloader, recorder| + reloader.reload_now + write("a.json", values_doc({ flag1: "fixed" })) + sleep 0.3 + + expect(recorder.applied).to be_empty + end + end + + it "coalesces triggers that arrive within the debounce delay into one reload" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0.2) do |reloader, recorder| + reloader.reload_now + write("a.json", values_doc({ flag1: "b" })) + 5.times { reloader.trigger } + + expect(wait_for { recorder.applied.length == 2 }).to be true + sleep 0.4 + expect(recorder.applied.length).to eq(2) + expect(recorder.flag_values).to eq({ flag1: "b" }) + end + end + + it "restarts the debounce window when a trigger arrives during it" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0.2) do |reloader, recorder| + reloader.reload_now + write("a.json", values_doc({ flag1: "b" })) + reloader.trigger + sleep 0.12 + reloader.trigger + sleep 0.12 + + expect(recorder.applied.length).to eq(1) + expect(wait_for { recorder.applied.length == 2 }).to be true + end + end + + it "reloads at once for each trigger when the debounce delay is zero" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0) do |reloader, recorder| + reloader.reload_now + write("a.json", values_doc({ flag1: "b" })) + reloader.trigger + expect(wait_for(0.5) { recorder.applied.length == 2 }).to be true + write("a.json", values_doc({ flag1: "c" })) + reloader.trigger + expect(wait_for(0.5) { recorder.applied.length == 3 }).to be true + expect(recorder.flag_values).to eq({ flag1: "c" }) + end + end + + it "applies a triggered reload even when the content is unchanged unless told to skip" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0) do |reloader, recorder| + reloader.reload_now + reloader.trigger + + expect(wait_for { recorder.applied.length == 2 }).to be true + end + end + + it "skips a reload whose content is identical to the last applied content" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0, skip_unchanged: true) do |reloader, recorder| + reloader.reload_now + reloader.trigger + sleep 0.3 + expect(recorder.applied.length).to eq(1) + + write("a.json", values_doc({ flag1: "b" })) + reloader.trigger + expect(wait_for { recorder.applied.length == 2 }).to be true + end + end + + it "applies a success that follows a failure even when the content is unchanged" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], retry_delay: 0, skip_unchanged: true) do |reloader, recorder| + reloader.reload_now + write("a.json", '{"flagValues"') + reloader.reload_now + write("a.json", values_doc({ flag1: "a" })) + reloader.reload_now + + expect(recorder.applied.length).to eq(2) + expect(recorder.errors.length).to eq(1) + end + end + + it "reports an exception from apply as a failed reload and applies the same content on the retry" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], reject: 1, retry_delay: 0.1) do |reloader, recorder| + expect(reloader.reload_now).to be false + expect(recorder.applied).to be_empty + expect(recorder.errors.length).to eq(1) + expect(recorder.errors[0]).to be_a(ArgumentError) + + # The retry applies the content that was rejected, so a consumer that recovers sees it. + expect(wait_for { recorder.applied.length == 1 }).to be true + expect(recorder.flag_values).to eq({ flag1: "a" }) + end + end + + it "does not skip content as unchanged after apply rejected it" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], reject: 1, retry_delay: 0, skip_unchanged: true) do |reloader, recorder| + expect(reloader.reload_now).to be false + + # The same content again. It was never applied, so it is not skipped as unchanged. + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag1: "a" }) + end + end + + it "reloads on a later trigger after apply raised during a triggered reload" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], reject: 1, debounce_delay: 0, retry_delay: 0) do |reloader, recorder| + reloader.trigger + expect(wait_for { recorder.errors.length == 1 }).to be true + + write("a.json", values_doc({ flag1: "b" })) + reloader.trigger + expect(wait_for { recorder.applied.length == 1 }).to be true + expect(recorder.flag_values).to eq({ flag1: "b" }) + end + end + + it "reports a repeated rejection by apply once" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], reject: 2, retry_delay: 0) do |reloader, recorder| + expect(reloader.reload_now).to be false + expect(reloader.reload_now).to be false + + # The second rejection repeats the first, so on_error is not invoked again. + expect(recorder.errors.length).to eq(1) + expect(reloader.reload_now).to be true + expect(recorder.applied.length).to eq(1) + end + end + + it "runs reloads one at a time even when triggers overlap" do + a = write("a.json", values_doc({ flag1: "a" })) + active = Concurrent::AtomicFixnum.new(0) + max_active = Concurrent::AtomicFixnum.new(0) + applied = Concurrent::AtomicFixnum.new(0) + apply = lambda do |_merged| + current = active.increment + max_active.update { |m| [m, current].max } + sleep 0.1 + active.decrement + applied.increment + end + reloader = Reloader.new(paths: [a], logger: $null_log, apply: apply, debounce_delay: 0) + + begin + threads = Array.new(4) do + Thread.new do + reloader.reload_now + reloader.trigger + end + end + threads.each(&:join) + + expect(wait_for { applied.value >= 5 }).to be true + expect(max_active.value).to eq(1) + ensure + reloader.stop + end + end + + it "does not apply after it is stopped" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_reloader([a], debounce_delay: 0) do |reloader, recorder| + reloader.reload_now + reloader.stop + write("a.json", values_doc({ flag1: "b" })) + reloader.trigger + expect(reloader.reload_now).to be true + sleep 0.2 + + expect(recorder.applied.length).to eq(1) + end + end + + it "does not report a failure after it is stopped" do + a = write("a.json", '{"flagValues"') + + with_reloader([a]) do |reloader, recorder| + reloader.stop + expect(reloader.reload_now).to be true + + expect(recorder.errors).to be_empty + end + end + + it "logs a change-triggered reload at info level" do + a = write("a.json", values_doc({ flag1: "a" })) + logger = CapturingLogger.new + + with_reloader([a], logger: logger, debounce_delay: 0) do |reloader, recorder| + reloader.reload_now + reloader.trigger + expect(wait_for { recorder.applied.length == 2 }).to be true + + expect(logger.output).to include("Reloading flag data after detecting a change") + end + end + + describe "a file that is found empty" do + # The retry comes long enough after the change signal for a test to finish writing the + # file before it, even on a slow machine. + let(:retry_delay) { 1.5 } + + # Shorter than the retry delay, so a load that happens within it after a change signal + # is not the retry. + let(:before_the_retry) { 1.0 } + + let(:logger) { CapturingLogger.new } + + let(:read_again_log) { "reading it again before applying it" } + + def yaml_doc(values) + "flagValues:\n" + values.map { |key, value| " #{key}: #{value}\n" }.join + end + + def read_again_logged? + logger.output.include?(read_again_log) + end + + def hold_count + logger.output.scan(read_again_log).length + end + + def with_empty_file_reloader(paths, **options, &block) + with_reloader(paths, logger: logger, debounce_delay: 0.01, retry_delay: retry_delay, **options, &block) + end + + it "is held back before it is applied" do + # A writer that rewrites the file in place truncates it and then writes the new content. + # In YAML an empty file is valid, so applying it would briefly remove every entry. + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + expect(logger.output).to include("DEBUG -- : [LDClient] File #{y} is empty, possibly while it is being rewritten") + + # The retry reads the rewritten file. + File.write(y, yaml_doc(flag2: true)) + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag2: true }) + expect(recorder.errors).to be_empty + expect(logger.output).not_to include("ERROR") + end + end + + it "counts a file with only white space as empty" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, " \n\r\n \n") + reloader.trigger + + expect(wait_for { read_again_logged? }).to be true + expect(recorder.applied.length).to eq(1) + end + end + + it "is applied by the retry when a YAML file stays empty" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + expect(wait_for(before_the_retry) { recorder.applied.length > 1 }).to be false + + # The file is still empty when the retry reads it, so it is a document with no entries. + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.applied[-1].empty?).to be true + expect(recorder.errors).to be_empty + end + end + + it "is not held back again once its empty state is applied" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + + # The applied data has no content from the file, so a new truncation is applied at once. + File.write(y, "") + reloader.trigger + expect(wait_for(before_the_retry) { recorder.applied.length == 3 }).to be true + expect(recorder.applied[-1].empty?).to be true + expect(logger.output.scan(read_again_log).length).to eq(1) + end + end + + it "is not reported by the load that finds an emptied JSON file, and the retry reports it" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_empty_file_reloader([a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + expect(recorder.errors).to be_empty + + # The retry finds the file still empty, which is not a JSON document. The data stays. + expect(wait_for(retry_delay + 2) { recorder.errors.length == 1 }).to be true + expect(recorder.errors[0]).to be_a(ReadError) + expect(recorder.errors[0].message).to include("the file has no content, which is not a JSON document") + expect(recorder.applied.length).to eq(1) + end + end + + it "is not applied by a further change signal within the retry delay" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + + # The file has not been written since, so the delay goes on from the first load that + # found it empty, and a stream of change signals cannot hold it back longer. + reloader.trigger + expect(wait_for { logger.output.scan(read_again_log).length == 2 }).to be true + expect(recorder.applied.length).to eq(1) + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "is held back again when a second rewrite truncates it before a load reads its content" do + # Two quick in-place rewrites: the write of the first and the truncation of the second + # come before the next load, which finds the file empty again. It is a new rewrite, so + # the delay starts again. + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + + sleep 0.02 + File.write(y, yaml_doc(flag2: true)) + File.write(y, "") + reloader.trigger + expect(wait_for { logger.output.scan(read_again_log).length == 2 }).to be true + + File.write(y, yaml_doc(flag3: true)) + reloader.trigger + expect(wait_for(before_the_retry) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag3: true }) + end + end + + it "is held back again when the retry finds a second truncation" do + # A retry, or the next poll, can find the file empty again after a second rewrite started. + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + + # No change signal from here on: only the retries read the file. + sleep 0.02 + File.write(y, yaml_doc(flag2: true)) + File.write(y, "") + expect(wait_for(retry_delay + 2) { logger.output.scan(read_again_log).length == 2 }).to be true + expect(recorder.applied.length).to eq(1) + + File.write(y, yaml_doc(flag3: true)) + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag3: true }) + end + end + + it "is held back again when a change to another file starts a load within the delay" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_empty_file_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + reloader.trigger + expect(wait_for { read_again_logged? }).to be true + + # The load for the change to b finds a still empty, within the delay. + File.write(b, yaml_doc(flag2: false)) + reloader.trigger + expect(wait_for { logger.output.scan(read_again_log).length == 2 }).to be true + expect(recorder.applied.length).to eq(1) + + File.write(a, yaml_doc(flag1: false)) + reloader.trigger + expect(wait_for(before_the_retry) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag1: false, flag2: false }) + end + end + + it "is held back after the same content moved to it from another file" do + # The same content moves from one file to another. The data in effect now comes from the + # second file, so the move is applied rather than skipped as unchanged, and emptying the + # second file is held back. + a = write("a.yaml", yaml_doc(flag1: true)) + b = path("b.yaml") + + with_empty_file_reloader([a, b], skip_missing_paths: true, skip_unchanged: true) do |reloader, recorder| + expect(reloader.reload_now).to be true + File.write(b, yaml_doc(flag1: true)) + File.delete(a) + expect(reloader.reload_now).to be true + expect(recorder.applied.length).to eq(2) + expect(recorder.applied[-1].files.map(&:present)).to eq([false, true]) + + File.write(b, "") + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(2) + end + end + + it "applies the same bytes moved to a file of another format" do + # 1e3 is a number in JSON and a string in YAML, so the move is not skipped as unchanged. + a = write("a.json", '{"flagValues": {"f": 1e3}}') + b = path("b.yaml") + + with_reloader([a, b], skip_missing_paths: true, skip_unchanged: true) do |reloader, recorder| + reloader.reload_now + expect(recorder.flag_values).to eq({ f: 1000.0 }) + + File.delete(a) + File.write(b, '{"flagValues": {"f": 1e3}}') + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ f: "1e3" }) + end + end + + it "does not keep a load from failing on another file that cannot be parsed" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.json", values_doc({ flag2: "b" })) + + with_empty_file_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + File.write(b, '{"flagValues"') + reloader.trigger + + # The load that puts a off reports b's failure, before the retry. + expect(wait_for(before_the_retry) { recorder.errors.length == 1 }).to be true + expect(recorder.errors[0].path).to eq(b) + expect(recorder.applied.length).to eq(1) + end + end + + it "does not delay the other files' changes when it is an empty placeholder from the start" do + # An empty YAML file next to the data, such as a placeholder for later entries, has never + # had content, so it is never read again. + a = write("a.json", values_doc({ flag1: "a" })) + placeholder = write("placeholder.yaml", "") + + with_empty_file_reloader([a, placeholder]) do |reloader, recorder| + reloader.reload_now + expect(recorder.flag_values).to eq({ flag1: "a" }) + + File.write(a, values_doc({ flag1: "b" })) + reloader.trigger + expect(wait_for(before_the_retry) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag1: "b" }) + expect(read_again_logged?).to be false + end + end + + it "is applied at once when it was missing at its last read" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y], skip_missing_paths: true) do |reloader, recorder| + reloader.reload_now + File.delete(y) + reloader.trigger + expect(wait_for { recorder.applied.length == 2 }).to be true + expect(recorder.applied[-1].empty?).to be true + + # A missing file has no content, so the empty file that replaces it is not read again. + File.write(y, "") + reloader.trigger + expect(wait_for(before_the_retry) { recorder.applied.length == 3 }).to be true + expect(recorder.applied[-1].empty?).to be true + expect(read_again_logged?).to be false + end + end + + it "is held back by a retry left over from an earlier failure instead of applied empty" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "flagValues: [") + reloader.trigger + expect(wait_for { recorder.errors.length == 1 }).to be true + + # No change signal from here on: only the retries read the file. + File.write(y, "") + expect(wait_for(retry_delay + 2) { read_again_logged? }).to be true + File.write(y, yaml_doc(flag2: true)) + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag2: true }) + end + end + + it "makes reload_now return false and arms the retry" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + expect(reloader.reload_now).to be true + File.write(y, "") + + expect(reloader.reload_now).to be false + expect(read_again_logged?).to be true + expect(recorder.errors).to be_empty + + # The retry applies the file, which is still empty. + expect(wait_for(retry_delay + 2) { recorder.applied.length == 2 }).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "neither logs nor arms a retry when the reloader is stopped while the files are read" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_empty_file_reloader([y]) do |reloader, recorder| + reloader.reload_now + allow(FileData).to receive(:read_file).and_wrap_original do |_original, _path| + reloader.stop + "" + end + + expect(reloader.reload_now).to be true + expect(read_again_logged?).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "is applied at once when there are no automatic retries" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_reloader([y], logger: logger, retry_delay: 0) do |reloader, recorder| + reloader.reload_now + + # Without a retry nothing would read the file again, so the load applies it as it is. + File.write(y, "") + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + expect(read_again_logged?).to be false + end + end + + it "is a document with no entries at the initial load when it is YAML" do + y = write("a.yaml", "") + + with_empty_file_reloader([y]) do |reloader, recorder| + expect(reloader.reload_now).to be true + expect(recorder.applied[0].empty?).to be true + end + end + + it "fails to parse at the initial load when it is JSON" do + a = write("a.json", "") + + with_empty_file_reloader([a]) do |reloader, recorder| + expect(reloader.reload_now).to be false + expect(recorder.applied).to be_empty + expect(recorder.errors[0].message).to include("the file has no content, which is not a JSON document") + end + end + + it "says that the retry after a hold reads the files again because a file was found empty" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_reloader([y], logger: logger, debounce_delay: 0.01, retry_delay: 0.2) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { recorder.applied.length == 2 }).to be true + expect(logger.output).to include("DEBUG -- : [LDClient] Reading flag data again after a file was found empty") + expect(logger.output).not_to include("Retrying flag data load after earlier failure") + + File.write(y, "flagValues: [") + reloader.trigger + expect(wait_for { logger.output.include?("Retrying flag data load after earlier failure") }).to be true + end + end + + it "keeps reloading after an error that a load does not expect" do + y = write("a.yaml", yaml_doc(flag1: true)) + failures = 1 + allow(FileData).to receive(:read_file).and_wrap_original do |original, path| + raise "unexpected" if (failures -= 1) >= 0 + + original.call(path) + end + + with_reloader([y], logger: logger, debounce_delay: 0.01, retry_delay: 0.2) do |reloader, recorder| + reloader.trigger + expect(wait_for { logger.output.include?("Unexpected error in file data reloader") }).to be true + + # The retry loads the file, and so does a later change signal. + expect(wait_for { recorder.applied.length == 1 }).to be true + File.write(y, yaml_doc(flag2: true)) + reloader.trigger + expect(wait_for { recorder.applied.length == 2 }).to be true + expect(recorder.flag_values).to eq({ flag2: true }) + end + end + + it "keeps the worker running after an error outside a load" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_reloader([y], logger: logger, debounce_delay: 0.01, retry_delay: 0.2) do |reloader, recorder| + failures = 1 + allow(reloader).to receive(:reload).and_wrap_original do |original, **options| + raise "unexpected" if (failures -= 1) >= 0 + + original.call(**options) + end + reloader.trigger + expect(wait_for { logger.output.include?("Unexpected error in file data reloader") }).to be true + + reloader.trigger + expect(wait_for { recorder.applied.length == 1 }).to be true + end + end + + it "replaces an earlier retry with the retry of a later hold" do + # A failure arms the retry for the retry delay. A load soon after holds a file back and + # arms its own retry for the end of that file's delay, which replaces the earlier one, + # so the file is not read again before its delay is over. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.json", values_doc({ flag2: "b" })) + + with_reloader([a, b], logger: logger, debounce_delay: 0.01, retry_delay: 1.0) do |reloader, recorder| + reloader.reload_now + File.write(b, '{"flagValues"') + reloader.trigger + expect(wait_for { recorder.errors.length == 1 }).to be true + + sleep 0.3 + File.write(a, "") + File.write(b, values_doc({ flag2: "b" })) + reloader.trigger + expect(wait_for { hold_count == 1 }).to be true + expect(wait_for(3) { recorder.applied.length == 2 }).to be true + expect(hold_count).to eq(1) + end + end + + it "arms the retry of a hold for the rest of the delay, plus a tenth of it" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_reloader([y], logger: logger, debounce_delay: 0.01, retry_delay: 2.0) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + expect(wait_for { hold_count == 1 }).to be true + first_hold = Time.now + + # A load halfway through the delay holds the file back again, and arms the retry for + # the rest of the delay, not for a whole delay. + sleep 1.0 + reloader.trigger + expect(wait_for { hold_count == 2 }).to be true + expect(wait_for(4) { recorder.applied.length == 2 }).to be true + expect(Time.now - first_hold).to be_between(1.9, 2.7) + end + end + end + + describe "holding back a file, on a controlled clock" do + # A monotonic clock that only moves when a test moves it. + class FakeClock + attr_reader :now + + def initialize + @now = 1000.0 + end + + def advance(seconds) + @now += seconds + end + end + + # A retry delay of an hour on the controlled clock, so that no real retry fires during a + # test. Each load is a synchronous reload_now. + let(:hour) { 3600.0 } + let(:clock) { FakeClock.new } + let(:logger) { CapturingLogger.new } + + def yaml_doc(values) + "flagValues:\n" + values.map { |key, value| " #{key}: #{value}\n" }.join + end + + def hold_count + logger.output.scan("reading it again before applying it").length + end + + # Lets the next write get a later modification time on a file system whose timestamps only + # move once per clock tick of the kernel. + def next_tick + sleep 0.02 + end + + # Waits for the worker to arm the retry that a load asked for, and expects its delay. A + # retry that reload_now asks for is armed by the worker. + def expect_armed_retry_delay(reloader, seconds) + armed = wait_for do + delay = reloader.last_armed_retry_delay + !delay.nil? && (delay - seconds).abs < 0.001 + end + expect(armed).to be(true), "expected a retry armed for #{seconds}s, got #{reloader.last_armed_retry_delay.inspect}" + end + + # Gives the file a new modification time, as a write while it is empty does. + def mark_written(path) + @marks = (@marks || 0) + 1 + File.utime(Time.at(@marks), Time.at(@marks), path) + end + + def with_clocked_reloader(paths, **options, &block) + with_reloader(paths, logger: logger, retry_delay: hour, clock: clock.method(:now), **options, &block) + end + + it "holds the file back until it has stayed empty for exactly the retry delay" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + + clock.advance(0.95 * hour) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + + clock.advance(0.05 * hour) + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "starts the delay over when the file is written while it is empty" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + + # Rewritten and truncated again between two loads. + clock.advance(0.6 * hour) + next_tick + File.write(y, yaml_doc(flag2: true)) + File.write(y, "") + expect(reloader.reload_now).to be false + + clock.advance(0.6 * hour) + expect(reloader.reload_now).to be false + + clock.advance(0.4 * hour) + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "reads the state of the file a symbolic link leads to" do + skip "symbolic links need extra privileges on Windows" if Gem.win_platform? + target = write("target.yaml", yaml_doc(flag1: true)) + link = path("link.yaml") + File.symlink(target, link) + + with_clocked_reloader([link]) do |reloader, recorder| + reloader.reload_now + File.write(target, "") + expect(reloader.reload_now).to be false + + clock.advance(0.6 * hour) + next_tick + File.write(target, yaml_doc(flag2: true)) + File.write(target, "") + expect(reloader.reload_now).to be false + + # The target was written, so its delay started again. + clock.advance(0.6 * hour) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "stops holding back after five retry delays while one file keeps being rewritten" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + + # Truncating the file again changes its modification time, so every load finds a new + # rewrite. + 9.times do + clock.advance(0.5 * hour) + next_tick + File.write(y, "") + expect(reloader.reload_now).to be false + end + + clock.advance(0.5 * hour) + next_tick + File.write(y, "") + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "stops holding back after five retry delays while two files are rewritten in turn" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + # Each load finds one of the files in the middle of a rewrite and the other one whole. + 9.times do |i| + clock.advance(0.5 * hour) + next_tick + emptied, rewritten = i.even? ? [b, a] : [a, b] + File.write(rewritten, yaml_doc(rewritten == a ? { flag1: true } : { flag2: true })) + File.write(emptied, "") + expect(reloader.reload_now).to be false + end + expect(recorder.applied.length).to eq(1) + + clock.advance(0.5 * hour) + File.write(b, yaml_doc(flag2: true)) + File.write(a, "") + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag2: true }) + end + end + + it "applies files that are emptied together after one retry delay" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + File.write(b, "") + expect(reloader.reload_now).to be false + expect(hold_count).to eq(2) + + clock.advance(hour) + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "holds back YAML that is only a comment or a document marker, and applies {} at once" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "# overrides for the incident\n") + expect(reloader.reload_now).to be false + File.write(y, "---\n") + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + + File.write(y, "{}\n") + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + end + end + + it "applies a JSON file that is {} at once" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_clocked_reloader([a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "{}") + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + expect(hold_count).to eq(0) + end + end + + it "keeps parsing a JSON file that stays empty, without holding it back again" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_clocked_reloader([a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + clock.advance(hour) + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + clock.advance(hour) + expect(reloader.reload_now).to be false + expect(hold_count).to eq(1) + expect(recorder.applied.length).to eq(1) + end + end + + it "holds back a JSON file that stayed empty for long when it is rewritten" do + # The failing loads held nothing back, so they started the file's bound over, and the + # rewrite is held back although the file was first found empty more than five retry + # delays ago. + a = write("a.json", values_doc({ flag1: "a" })) + + with_clocked_reloader([a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + clock.advance(hour) + expect(reloader.reload_now).to be false + clock.advance(5 * hour) + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + + next_tick + File.write(a, "") + expect(reloader.reload_now).to be false + expect(hold_count).to eq(2) + File.write(a, values_doc({ flag1: "b" })) + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag1: "b" }) + expect(recorder.errors.length).to eq(1) + end + end + + it "holds back a file emptied again after its content was read, even if its state looks the same" do + # On a file system with coarse modification times, a rewrite can leave the state of an + # empty file unchanged. + y = write("a.yaml", yaml_doc(flag1: true)) + allow(Poller).to receive(:observe).and_return(Poller::FileState.new(true, Time.at(0), 0)) + + with_clocked_reloader([y]) do |reloader, _recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + File.write(y, yaml_doc(flag2: true)) + expect(reloader.reload_now).to be true + + clock.advance(2 * hour) + File.write(y, "") + expect(reloader.reload_now).to be false + end + end + + it "starts a file's delay again after a load reads it with content, even if its state looks the same" do + # The load that reads a.yaml with content holds b.yaml back, so nothing is applied in between. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + allow(Poller).to receive(:observe).and_return(Poller::FileState.new(true, Time.at(0), 0)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + clock.advance(0.5 * hour) + File.write(a, yaml_doc(flag1: true)) + File.write(b, "") + expect(reloader.reload_now).to be false + + clock.advance(0.1 * hour) + File.write(a, "") + File.write(b, yaml_doc(flag2: true)) + expect(reloader.reload_now).to be false + + # a.yaml's delay started again at 0.6 hours, not at the start. + clock.advance(0.6 * hour) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "holds a file emptied again after its content was read back again when its state looks the same" do + # On a file system that records modification times coarsely, a file emptied a second time + # can look exactly as it did the first time. The read of its content in between still + # starts its delay over. b.yaml is kept written while empty, so that no load applies. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + first_empty_write_time = File.mtime(a) + File.write(b, "") + expect(reloader.reload_now).to be false + + clock.advance(50.0 / 60 * hour) + File.write(a, yaml_doc(flag1: true)) + mark_written(b) + expect(reloader.reload_now).to be false + + clock.advance(20.0 / 60 * hour) + File.write(a, "") + File.utime(first_empty_write_time, first_empty_write_time, a) + mark_written(b) + expect(reloader.reload_now).to be false + + # b.yaml gets its content back while a.yaml is still within its new delay. + clock.advance(10.0 / 60 * hour) + File.write(b, yaml_doc(flag2: true)) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "holds back a file created empty after it was missing, even if its state looks the same" do + y = write("a.yaml", yaml_doc(flag1: true)) + allow(Poller).to receive(:observe).and_return(Poller::FileState.new(true, Time.at(0), 0)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + + # Without skipping missing paths, a missing file fails the load and the data stays. + File.delete(y) + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + + clock.advance(2 * hour) + File.write(y, "") + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "does not let a slow read of an earlier file shorten a later file's delay" do + # The read of b.yaml, which comes first, takes most of a retry delay. a.yaml's delay is + # counted from when it was read. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + slow = nil + allow(FileData).to receive(:read_file).and_wrap_original do |original, path| + content = original.call(path) + if path == slow + slow = nil + clock.advance(0.9 * hour) + end + content + end + + with_clocked_reloader([b, a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + slow = b + expect(reloader.reload_now).to be false + + clock.advance(0.2 * hour) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "does not start the bound of a held file over when another file fails to load" do + a = write("a.yaml", yaml_doc(flag1: true)) + c = write("c.json", values_doc({ flag3: "c" })) + + with_clocked_reloader([a, c]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + # a.yaml keeps being written while empty, and every other load fails on c.json. + 9.times do |i| + clock.advance(0.5 * hour) + mark_written(a) + File.write(c, i.even? ? '{"flagValues"' : values_doc({ flag3: "c" })) + expect(reloader.reload_now).to be false + end + expect(recorder.applied.length).to eq(1) + + clock.advance(0.5 * hour) + mark_written(a) + File.write(c, values_doc({ flag3: "c" })) + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag3: "c" }) + end + end + + it "does not cut a file's rewrite short when another file reaches its bound" do + # b.yaml keeps being written while empty, and reaches its bound while a.yaml is in the + # middle of an ordinary rewrite. a.yaml is still held back. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(b, "") + expect(reloader.reload_now).to be false + 9.times do |i| + clock.advance(0.5 * hour) + mark_written(b) + File.write(a, "") if i == 8 + expect(reloader.reload_now).to be false + end + + clock.advance(0.5 * hour) + mark_written(b) + expect(reloader.reload_now).to be false + + clock.advance(0.25 * hour) + File.write(a, yaml_doc(flag3: true)) + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag3: true }) + end + end + + it "starts the bound over once data is applied" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + 8.times do + clock.advance(0.5 * hour) + mark_written(y) + expect(reloader.reload_now).to be false + end + File.write(y, yaml_doc(flag2: true)) + expect(reloader.reload_now).to be true + + # Emptied again: a new bound of five retry delays, not the rest of the old one, which + # would be over at 5 hours. + clock.advance(0.1 * hour) + File.write(y, "") + expect(reloader.reload_now).to be false + 2.times do + clock.advance(0.5 * hour) + mark_written(y) + expect(reloader.reload_now).to be false + end + expect(recorder.flag_values).to eq({ flag2: true }) + end + end + + it "holds back a file again later after a rewrite with the same content was skipped as unchanged" do + # A tool rewrites the file in place now and then with the same content. The load that + # finds the content again is skipped as unchanged, and starts the file's bound over, as + # an application does. + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y], skip_unchanged: true) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + File.write(y, yaml_doc(flag1: true)) + expect(reloader.reload_now).to be true + expect(recorder.applied.length).to eq(1) + + clock.advance(6 * hour) + File.write(y, "") + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "applies files emptied in staggered turns after the bound" do + # Two files stay empty and are truncated again in turn, so that whenever one of them has + # stayed empty for the whole delay, the other is still within its delay and holds the + # load back. A file that settles so does not get a fresh bound while the other is held. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + # One file is truncated again every 0.9 retry delays, the files in turn, with a load at + # each truncation and one 0.2 retry delays later, which stands for the retry that the + # hold arms a little after the other file's delay is over. + applied = false + (1..14).each do |i| + clock.advance(i == 1 ? 0.9 * hour : 0.7 * hour) + mark_written(i.odd? ? b : a) + File.write(b, "") if i == 1 + reloader.reload_now + clock.advance(0.2 * hour) + reloader.reload_now + applied = recorder.applied.length > 1 + break if applied + end + expect(applied).to be true + end + end + + it "starts the bound of a file held back during a failure over once a load reads it again" do + # While b.json fails to load, a.yaml is rewritten once and its gap is held back. Later + # b.json is fixed while a.yaml is in the middle of another rewrite, which is held back + # with a bound of its own. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.json", values_doc({ flag3: "c" })) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(b, '{"flagValues"') + File.write(a, "") + expect(reloader.reload_now).to be false + File.write(a, yaml_doc(flag2: true)) + expect(reloader.reload_now).to be false + clock.advance(6 * hour) + expect(reloader.reload_now).to be false + + # The load that held a.yaml back first failed on b.json, so that hold was not logged. + File.write(b, values_doc({ flag3: "c" })) + File.write(a, "") + expect(reloader.reload_now).to be false + expect(hold_count).to eq(1) + expect(recorder.applied.length).to eq(1) + end + end + + it "asks for the retry from the end of the load's reads" do + # The read of the file after the held one takes longer than what is left of the held + # file's delay, so the retry is due right away, after the margin. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + slow = nil + allow(FileData).to receive(:read_file).and_wrap_original do |original, path| + content = original.call(path) + if path == slow + slow = nil + clock.advance(0.1 * hour) + end + content + end + + with_clocked_reloader([a, b]) do |reloader, _recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + clock.advance(0.95 * hour) + slow = b + expect(reloader.reload_now).to be false + expect_armed_retry_delay(reloader, 0.1 * hour) + end + end + + [ + ["YAML", "a.yaml", "flagValues:\n flag1: true\n", "flagValues: [\n", "flagValues:\n flag2: true\n"], + ["JSON", "a.json", '{"flagValues": {"flag1": true}}', '{"flagValues": ', '{"flagValues": {"flag2": true}}'], + ].each do |format, name, good, broken, fixed| + it "holds back the in-place save that fixes a #{format} file saved with an error" do + # An operator saves the file in place with a mistake, and saves the fix in place long + # after. The loads in between fail on the file, and the fix's own gap is held back. + y = write(name, good) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + File.write(y, broken) + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + 6.times do + clock.advance(hour) + reloader.reload_now + end + errors = logger.output.scan("ERROR").length + + File.write(y, "") + expect(reloader.reload_now).to be false + expect(hold_count).to eq(2) + expect(logger.output.scan("ERROR").length).to eq(errors) + File.write(y, fixed) + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag2: true }) + end + end + end + + it "holds back the fix of a file saved with an error even when its emptying looks like the first" do + # The broken content ended the first empty state, so the later truncation starts a new + # delay although, on a file system with coarse modification times, it looks the same. + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, _recorder| + reloader.reload_now + File.write(y, "") + first_empty_write_time = File.mtime(y) + expect(reloader.reload_now).to be false + File.write(y, "flagValues: [\n") + expect(reloader.reload_now).to be false + + clock.advance(2 * hour) + File.write(y, "") + File.utime(first_empty_write_time, first_empty_write_time, y) + expect(reloader.reload_now).to be false + expect(hold_count).to eq(2) + end + end + + it "starts the bound over for a file whose read fails at a load that holds nothing back" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + + allow(FileData).to receive(:read_file).and_raise(ReadError.new(y, "unable to read file: denied")) + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + allow(FileData).to receive(:read_file).and_call_original + + clock.advance(6 * hour) + mark_written(y) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "starts the bound over for a file found missing at a load that holds nothing back" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.json", values_doc({ flag3: "c" })) + + with_clocked_reloader([a, b], skip_missing_paths: true) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + + File.delete(a) + File.write(b, '{"flagValues"') + expect(reloader.reload_now).to be false + + # b.json is fixed while a.yaml is created empty again, in the middle of a rewrite. + clock.advance(6 * hour) + File.write(a, "") + File.write(b, values_doc({ flag3: "c" })) + expect(reloader.reload_now).to be false + expect(hold_count).to eq(2) + expect(recorder.applied.length).to eq(1) + end + end + + it "applies a file kept written while empty after its bound while an earlier file fails" do + # c.json comes first and fails now and then, while a.yaml keeps being written while empty. + # Those loads hold a.yaml back, so they do not give it a fresh bound. + c = write("c.json", values_doc({ flag3: "c" })) + a = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([c, a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + (1..9).each do |i| + clock.advance(0.5 * hour) + mark_written(a) + File.write(c, i.odd? ? '{"flagValues"' : values_doc({ flag3: "c" })) + reloader.reload_now + end + + clock.advance(0.5 * hour) + mark_written(a) + File.write(c, values_doc({ flag3: "c" })) + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag3: "c" }) + end + end + + it "starts the bound of a file after a failing file over once a load reads it again" do + # c.json comes first and is saved with an error while a.yaml is being rewritten. The loads + # that fail on c.json still read a.yaml, with content, so its bound starts over, and its + # next rewrite, when c.json is fixed, is held back. + c = write("c.json", values_doc({ flag3: "c" })) + a = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([c, a]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + File.write(c, '{"flagValues"') + File.write(a, yaml_doc(flag1: false)) + expect(reloader.reload_now).to be false + 6.times do + clock.advance(hour) + reloader.reload_now + end + + File.write(c, values_doc({ flag3: "c" })) + File.write(a, "") + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "fails a load with the first failure when several files fail" do + a = write("a.json", values_doc({ flag1: "a" })) + b = write("b.json", values_doc({ flag2: "b" })) + + with_clocked_reloader([a, b]) do |reloader, recorder| + reloader.reload_now + File.write(a, '{"flagValues"') + File.write(b, "[1") + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + expect(recorder.errors[0].path).to eq(a) + end + end + + it "does not start bounds over at a load that holds a file back and then fails" do + # a.yaml and b.yaml are rewritten in turn, so that every load holds one of them back, and + # c.json fails at every other load. Those loads hold a file back, so they start no bound + # over. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + c = write("c.json", values_doc({ flag3: "c" })) + + with_clocked_reloader([a, b, c]) do |reloader, recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + applied = false + (1..12).each do |i| + clock.advance(0.5 * hour) + rewritten, emptied = i.odd? ? [a, b] : [b, a] + File.write(rewritten, yaml_doc(rewritten == a ? { flag1: true } : { flag2: true })) + File.write(emptied, "") + File.write(c, i.odd? ? '{"flagValues"' : values_doc({ flag3: "c" })) + reloader.reload_now + applied = recorder.applied.length > 1 + break if applied + end + expect(applied).to be true + end + end + + it "starts a new delay for a file that is created empty again after it was missing" do + # While b.json fails, a.yaml is emptied, deleted, and created empty again with the state it + # had when it was first found empty, as within one tick of a coarse clock. Being missing + # ended its empty state, so it is held back with a new delay. + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.json", values_doc({ flag3: "c" })) + + with_clocked_reloader([a, b], skip_missing_paths: true) do |reloader, recorder| + reloader.reload_now + File.write(b, '{"flagValues"') + File.write(a, "") + first_empty_write_time = File.mtime(a) + reloader.reload_now + clock.advance(0.5 * hour) + File.delete(a) + reloader.reload_now + clock.advance(40.0 / 60 * hour) + File.write(a, "") + File.utime(first_empty_write_time, first_empty_write_time, a) + reloader.reload_now + + File.write(b, values_doc({ flag3: "c" })) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "holds back YAML with only a tab and then applies it as no entries" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "\t\n") + expect(reloader.reload_now).to be false + + clock.advance(hour) + expect(reloader.reload_now).to be true + expect(recorder.applied[-1].empty?).to be true + expect(recorder.errors).to be_empty + end + end + + it "starts the delay over when a YAML file with no document changes size but not modification time" do + # Partial reads of a file being written can find only a comment, of different lengths, on + # a file system with coarse modification times. + y = write("a.yaml", yaml_doc(flag1: true)) + pinned = Time.at(1_000_000) + + with_clocked_reloader([y]) do |reloader, recorder| + reloader.reload_now + File.write(y, "# one\n") + File.utime(pinned, pinned, y) + expect(reloader.reload_now).to be false + + clock.advance(0.6 * hour) + File.write(y, "# three\n") + File.utime(pinned, pinned, y) + expect(reloader.reload_now).to be false + + clock.advance(0.6 * hour) + expect(reloader.reload_now).to be false + expect(recorder.applied.length).to eq(1) + end + end + + it "does not start the bound of a file that it does not hold back" do + # a.json stays empty past its delay while b.yaml holds every load back. Those loads do not + # hold a.json back, so they give it no bound, and its later rewrite gets a full delay. + b = write("b.yaml", yaml_doc(flag2: true)) + a = write("a.json", values_doc({ flag1: "a" })) + + with_clocked_reloader([b, a]) do |reloader, _recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + clock.advance(hour) + expect(reloader.reload_now).to be false + + clock.advance(0.5 * hour) + File.write(b, "") + expect(reloader.reload_now).to be false + 9.times do + clock.advance(0.5 * hour) + mark_written(b) + expect(reloader.reload_now).to be false + end + + # b.yaml's bound is over, and a.json is truncated again. + clock.advance(0.5 * hour) + mark_written(b) + mark_written(a) + holds = hold_count + expect(reloader.reload_now).to be false + expect(hold_count).to eq(holds + 1) + end + end + + it "asks for the retry when the delay is over, plus a tenth of it" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, _recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + expect_armed_retry_delay(reloader, 1.1 * hour) + + # A later load asks for the rest of the delay, not a whole one. + clock.advance(0.5 * hour) + expect(reloader.reload_now).to be false + expect_armed_retry_delay(reloader, 0.6 * hour) + end + end + + it "asks for the retry when the bound is over, if that comes first" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y]) do |reloader, _recorder| + reloader.reload_now + File.write(y, "") + expect(reloader.reload_now).to be false + 4.times do + clock.advance(hour) + mark_written(y) + expect(reloader.reload_now).to be false + end + clock.advance(0.8 * hour) + mark_written(y) + expect(reloader.reload_now).to be false + + # The file's delay would be over in an hour, but its bound is over in 0.2 hours. + expect_armed_retry_delay(reloader, 0.3 * hour) + end + end + + it "asks for the retry when the first of several held files is due" do + a = write("a.yaml", yaml_doc(flag1: true)) + b = write("b.yaml", yaml_doc(flag2: true)) + + with_clocked_reloader([a, b]) do |reloader, _recorder| + reloader.reload_now + File.write(a, "") + expect(reloader.reload_now).to be false + clock.advance(0.25 * hour) + File.write(b, "") + expect(reloader.reload_now).to be false + + expect_armed_retry_delay(reloader, 0.85 * hour) + end + end + + it "asks for the same retry when a change signal starts the load that holds a file back" do + y = write("a.yaml", yaml_doc(flag1: true)) + + with_clocked_reloader([y], debounce_delay: 0.01) do |reloader, _recorder| + reloader.reload_now + File.write(y, "") + reloader.trigger + + expect(wait_for { hold_count == 1 }).to be true + expect_armed_retry_delay(reloader, 1.1 * hour) + end + end + + it "asks for the retry delay after a failure" do + a = write("a.json", values_doc({ flag1: "a" })) + + with_clocked_reloader([a]) do |reloader, recorder| + reloader.reload_now + File.write(a, '{"flagValues"') + expect(reloader.reload_now).to be false + expect(recorder.errors.length).to eq(1) + expect_armed_retry_delay(reloader, hour) + end + end + + it "does not hold back a placeholder or a file that was missing" do + a = write("a.yaml", yaml_doc(flag1: true)) + placeholder = write("placeholder.yaml", "") + later = path("later.yaml") + + with_clocked_reloader([a, placeholder, later], skip_missing_paths: true) do |reloader, recorder| + reloader.reload_now + File.write(later, "") + expect(reloader.reload_now).to be true + expect(recorder.flag_values).to eq({ flag1: true }) + expect(hold_count).to eq(0) + end + end + end + end + end + end +end diff --git a/spec/impl/file_data/watcher_spec.rb b/spec/impl/file_data/watcher_spec.rb new file mode 100644 index 00000000..d5357456 --- /dev/null +++ b/spec/impl/file_data/watcher_spec.rb @@ -0,0 +1,307 @@ +# frozen_string_literal: true + +require "spec_helper" +require "capturing_logger" +require "fileutils" +require "tmpdir" +require "ldclient-rb/impl/file_data" + +module LaunchDarkly + module Impl + module FileData + describe Watcher do + before do + skip "the listen gem is not installed" unless Watcher.available? + end + + around do |example| + Dir.mktmpdir do |dir| + @dir = dir + example.run + end + end + + def path(name) + File.join(@dir, name) + end + + def wait_for(timeout = 5) + deadline = Time.now + timeout + until yield + return false if Time.now > deadline + sleep 0.02 + end + true + end + + def with_watcher(paths, logger: $null_log) + calls = Concurrent::AtomicFixnum.new(0) + watcher = Watcher.new(paths, -> { calls.increment }, logger) + begin + yield watcher, calls + ensure + watcher.stop + end + end + + # Returns the call count once it has stopped changing for a moment. One edit can produce a + # burst of notifications, and a count taken in the middle of the burst would make the rest + # of it look like a later signal. + def settled(calls) + count = calls.value + loop do + sleep 0.2 + break if calls.value == count + count = calls.value + end + count + end + + it "reports that the listen gem is available" do + expect(Watcher.available?).to be true + end + + it "invokes the callback when a watched file is modified" do + File.write(path("a.json"), "{}") + with_watcher([path("a.json")]) do |_watcher, calls| + sleep 0.3 + File.write(path("a.json"), '{"flagValues": {}}') + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "invokes the callback when a watched file that did not exist appears" do + with_watcher([path("later.json")]) do |_watcher, calls| + sleep 0.3 + File.write(path("later.json"), "{}") + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "invokes the callback when a watched file is deleted" do + File.write(path("a.json"), "{}") + with_watcher([path("a.json")]) do |_watcher, calls| + sleep 0.3 + File.delete(path("a.json")) + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "ignores other files in the same directory" do + File.write(path("a.json"), "{}") + with_watcher([path("a.json")]) do |_watcher, calls| + sleep 0.3 + File.write(path("other.json"), "{}") + sleep 0.5 + expect(calls.value).to eq(0) + end + end + + it "watches files in more than one directory" do + Dir.mkdir(path("sub")) + File.write(path("a.json"), "{}") + File.write(path("sub/b.json"), "{}") + with_watcher([path("a.json"), path("sub/b.json")]) do |_watcher, calls| + sleep 0.3 + File.write(path("sub/b.json"), '{"flagValues": {}}') + expect(wait_for { calls.value >= 1 }).to be true + end + end + + it "logs when a directory does not exist and starts watching once it appears" do + logger = CapturingLogger.new + missing_dir = path("not-yet") + with_watcher([File.join(missing_dir, "a.json")], logger: logger) do |_watcher, calls| + expect(logger.output).to include("Unable to watch data files") + expect(logger.output).to include("directory does not exist") + sleep 0.2 + expect(calls.value).to eq(0) + + Dir.mkdir(missing_dir) + # The watches are set up on the next retry, and the callback runs once at that point. + expect(wait_for { calls.value >= 1 }).to be true + before = calls.value + + sleep 0.3 + File.write(File.join(missing_dir, "a.json"), "{}") + expect(wait_for { calls.value > before }).to be true + end + end + + it "logs when a watched directory is deleted and watches it again once it exists again" do + skip "rb-inotify is not available on this platform" unless Watcher.inotify_available? + + logger = CapturingLogger.new + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + with_watcher([path("sub/a.json")], logger: logger) do |_watcher, calls| + sleep 0.3 + FileUtils.rm_rf(path("sub")) + expect(wait_for { logger.output.include?("WARN") }).to be true + expect(logger.output).to match(/WARN.*#{Regexp.escape(path('sub'))}/) + before = calls.value + + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + # The watches are set up on the next retry, and the callback runs once at that point. + expect(wait_for { calls.value > before }).to be true + before = calls.value + + sleep 0.3 + File.write(path("sub/a.json"), '{"flagValues": {}}') + expect(wait_for { calls.value > before }).to be true + end + end + + it "keeps retrying when a directory is lost again as soon as its watches are set up" do + skip "rb-inotify is not available on this platform" unless Watcher.inotify_available? + + logger = CapturingLogger.new + with_watcher([path("sub/a.json")], logger: logger) do |watcher, calls| + # Take the directory away as soon as the watches are set up, and wait until the loss + # has been handled, so that it is reported while the retry that set up the watches + # is still finishing. + lost_once = false + allow(watcher).to receive(:try_start).and_wrap_original do |original| + started = original.call + if started && !lost_once + lost_once = true + FileUtils.rm_rf(path("sub")) + # The loss has been handled once it is logged. The inotify thread cannot mark it, + # as it used to: each directory has its own watch now, and the thread outlives one. + wait_for { logger.output.include?("no longer exists") } + end + started + end + + Dir.mkdir(path("sub")) + expect(wait_for { calls.value >= 1 }).to be true + + # The retry that follows the second loss sets the watches up once the directory exists. + Dir.mkdir(path("sub")) + expect(wait_for { calls.value >= 2 }).to be true + end + end + + it "watches a directory that contains an unreadable subdirectory" do + skip "the current user can read every directory" if Process.uid.zero? + + Dir.mkdir(path("private")) + File.chmod(0o000, path("private")) + begin + File.write(path("a.json"), "{}") + logger = CapturingLogger.new + with_watcher([path("a.json")], logger: logger) do |_watcher, calls| + expect(logger.output).not_to include("Unable to watch data files") + sleep 0.3 + File.write(path("a.json"), '{"flagValues": {}}') + expect(wait_for { calls.value >= 1 }).to be true + end + ensure + File.chmod(0o755, path("private")) + end + end + + it "uses inotify on Linux and runs it on a named thread that stop ends" do + skip "rb-inotify is not available on this platform" unless Watcher.inotify_available? + + File.write(path("a.json"), "{}") + watcher = Watcher.new([path("a.json")], -> {}, $null_log) + threads = Thread.list.select { |t| t.name == "LD/FileDataWatcher" } + expect(threads.length).to eq 1 + + watcher.stop + + expect(threads[0].alive?).to be false + end + + it "does not invoke the callback after it is stopped" do + File.write(path("a.json"), "{}") + with_watcher([path("a.json")]) do |watcher, calls| + sleep 0.3 + watcher.stop + File.write(path("a.json"), '{"flagValues": {}}') + sleep 0.5 + expect(calls.value).to eq(0) + end + end + + it "can be stopped while it is still retrying a missing directory" do + with_watcher([path("not-yet/a.json")]) do |watcher, _calls| + watcher.stop + expect(Thread.list.map(&:name)).not_to include("LD/FileDataWatcherRetry") + end + end + + it "watches the directories that exist while another one is missing" do + logger = CapturingLogger.new + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + missing_dir = path("not-yet") + with_watcher([path("sub/a.json"), File.join(missing_dir, "b.json")], logger: logger) do |_watcher, calls| + expect(logger.output).to include("directory does not exist: #{missing_dir}") + sleep 0.3 + # The directory that exists is watched at once, not only when every directory exists. + File.write(path("sub/a.json"), '{"flagValues": {}}') + expect(wait_for { calls.value >= 1 }).to be true + before = settled(calls) + + Dir.mkdir(missing_dir) + # The watch is set up on the next retry, and the callback runs once at that point. + expect(wait_for { calls.value > before }).to be true + # Nothing is missing any more, so the retry ends and does not signal again. + expect(wait_for { Thread.list.none? { |t| t.name == "LD/FileDataWatcherRetry" } }).to be true + before = settled(calls) + sleep 0.3 + expect(calls.value).to eq(before) + + File.write(File.join(missing_dir, "b.json"), "{}") + expect(wait_for { calls.value > before }).to be true + end + end + + it "watches a lost directory again while another directory is still missing" do + skip "rb-inotify is not available on this platform" unless Watcher.inotify_available? + + logger = CapturingLogger.new + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + with_watcher([path("sub/a.json"), path("not-yet/b.json")], logger: logger) do |_watcher, calls| + FileUtils.rm_rf(path("sub")) + expect(wait_for { logger.output.match?(/WARN.*#{Regexp.escape(path('sub'))}/) }).to be true + before = settled(calls) + + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + # The directory is watched again on the next retry, which signals once at that point, + # although the other directory is still missing. + expect(wait_for { calls.value > before }).to be true + before = settled(calls) + + File.write(path("sub/a.json"), '{"flagValues": {}}') + expect(wait_for { calls.value > before }).to be true + expect(Dir.exist?(path("not-yet"))).to be false + expect(logger.output).to include("directory does not exist: #{path('not-yet')}") + end + end + + it "stops while a directory is missing and runs no callback afterwards" do + Dir.mkdir(path("sub")) + File.write(path("sub/a.json"), "{}") + with_watcher([path("sub/a.json"), path("not-yet/b.json")]) do |watcher, calls| + sleep 0.3 + watcher.stop + expect(Thread.list.map(&:name)).not_to include("LD/FileDataWatcherRetry") + expect(Thread.list.map(&:name)).not_to include("LD/FileDataWatcher") if Watcher.inotify_available? + + Dir.mkdir(path("not-yet")) + File.write(path("not-yet/b.json"), "{}") + File.write(path("sub/a.json"), '{"flagValues": {}}') + sleep 0.5 + expect(calls.value).to eq(0) + end + end + end + end + end +end diff --git a/spec/in_place_writes.rb b/spec/in_place_writes.rb new file mode 100644 index 00000000..bfba8546 --- /dev/null +++ b/spec/in_place_writes.rb @@ -0,0 +1,29 @@ +# +# Rewrites files the way applications and editors often do: in place, which is not atomic. The file +# is truncated first, and the new content is written after a pause, so that a reader in between +# finds the file empty, as it can with an application's writes, instead of only by chance. +# +module InPlaceWrites + # + # Truncates the file, writes the first part of the content if one is given, waits for the pause, + # then writes the rest. A file that does not exist is created first. + # + # @param path [String] + # @param content [String] + # @param pause [Numeric] seconds between truncating the file and writing the rest of the content + # @param first [String] the beginning of the content, which a reader can see during the pause + # + def self.rewrite(path, content, pause, first: "") + raise ArgumentError, "the content must start with the first part" unless content.start_with?(first) + + # Opening with "w" truncates the file at once. + File.open(path, "w") do |file| + unless first.empty? + file.write(first) + file.flush + end + sleep pause + file.write(content[first.length..]) + end + end +end diff --git a/spec/integrations/file_data_source_compatibility_spec.rb b/spec/integrations/file_data_source_compatibility_spec.rb new file mode 100644 index 00000000..5ba5a65b --- /dev/null +++ b/spec/integrations/file_data_source_compatibility_spec.rb @@ -0,0 +1,423 @@ +require "spec_helper" +require "capturing_logger" +require "tempfile" +require "ldclient-rb/integrations/file_data" + +# +# These specs pin behaviors of the existing file data sources that their other specs do not +# assert. The file-based override source shares the document format but not the implementation, +# and these sources must keep behaving as they always have. +# +module LaunchDarkly + module Integrations + describe "file data source behavior" do + let(:logger) { CapturingLogger.new } + + before do + @tmp_dir = Dir.mktmpdir + end + + after do + FileUtils.rm_rf(@tmp_dir) + end + + def make_temp_file(content) + file = Tempfile.new('flags', @tmp_dir) + IO.write(file, content) + file + end + + # Moves a file's modification time into the past. The pollers compare modification times, + # so a write that follows at once registers as a change even where the file system records + # the time in whole seconds. + def backdate(file) + past = Time.now - 10 + File.utime(past, past, file.path) + end + + def wait_for(timeout = 5) + deadline = Time.now + timeout + until yield + return false if Time.now > deadline + sleep 0.05 + end + true + end + + # Returns the next item from the queue, or nil when none arrives within the timeout. + # Queue#pop accepts a timeout only from Ruby 3.2 on, and the gem supports earlier runtimes. + def pop_with_timeout(queue, timeout) + wait_for(timeout) { !queue.empty? } ? queue.pop : nil + end + + def flag_json(key, version: nil) + data = { key: key, on: true, fallthrough: { variation: 0 }, variations: ["a"] } + data[:version] = version unless version.nil? + data + end + + describe FileData, "data_source" do + # Counts every init call so that a test can see each reload. + class CountingFeatureStore < InMemoryFeatureStore + attr_reader :init_count + + def initialize + super + @init_count = 0 + end + + def init(all_data) + @init_count += 1 + super + end + end + + before do + @store = CountingFeatureStore.new + @config = LaunchDarkly::Config.new(logger: logger, feature_store: @store) + executor = SynchronousExecutor.new + @status_broadcaster = LaunchDarkly::Impl::Broadcaster.new(executor, logger) + @flag_change_broadcaster = LaunchDarkly::Impl::Broadcaster.new(executor, logger) + @config.data_source_update_sink = LaunchDarkly::Impl::DataSource::UpdateSink.new(@store, @status_broadcaster, @flag_change_broadcaster) + end + + def with_data_source(options) + ds = FileData.data_source(options).call('', @config) + begin + yield ds + ensure + ds.stop + end + end + + def flag_version(key) + @store.get(Impl::DataStore::FEATURES, key).version + end + + def status + @config.data_source_update_sink.current_status + end + + it "numbers versions per file in load order and keeps counting across reloads" do + file1 = make_temp_file({ flags: { flag1: flag_json("flag1", version: 99) }, flagValues: { value1: true } }.to_json) + file2 = make_temp_file({ segments: { seg1: { key: "seg1" } } }.to_json) + backdate(file2) + + with_data_source({ paths: [file1.path, file2.path], auto_update: true, force_polling: true, poll_interval: 0.1 }) do |ds| + ds.start + expect(flag_version("flag1")).to eq 1 + expect(flag_version("value1")).to eq 1 + expect(@store.get(Impl::DataStore::SEGMENTS, "seg1").version).to eq 2 + + IO.write(file2, { segments: { seg1: { key: "seg1" }, seg2: { key: "seg2" } } }.to_json) + expect(wait_for { @store.get(Impl::DataStore::SEGMENTS, "seg2") }).to be true + expect(flag_version("flag1")).to eq 3 + expect(@store.get(Impl::DataStore::SEGMENTS, "seg2").version).to eq 4 + end + end + + it "stores an item under its own key member rather than the key it appears under" do + file = make_temp_file({ flags: { "map-key": flag_json("own-key") } }.to_json) + + with_data_source({ paths: [file.path] }) do |ds| + ds.start + expect(@store.get(Impl::DataStore::FEATURES, "own-key")).not_to be_nil + expect(@store.get(Impl::DataStore::FEATURES, "map-key")).to be_nil + end + end + + it "expands a flag value into a flag that is on and serves the value through its fallthrough" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + with_data_source({ paths: [file.path] }) do |ds| + ds.start + flag = @store.get(Impl::DataStore::FEATURES, "value1") + expect(flag.on).to be true + expect(flag.fallthrough.variation).to eq 0 + expect(flag.off_variation).to be_nil + expect(flag.variations).to eq ["x"] + end + end + + it "does not retry a failed initial load on its own" do + file = make_temp_file('{"flagValues"') + + with_data_source({ paths: [file.path] }) do |ds| + ds.start + expect(ds.initialized?).to be false + IO.write(file, { flagValues: { value1: "x" } }.to_json) + sleep 1.5 + + expect(ds.initialized?).to be false + expect(@store.init_count).to eq 0 + end + end + + it "logs a failed load with the file path at error level" do + file = make_temp_file('{"flagValues"') + + with_data_source({ paths: [file.path] }) do |ds| + ds.start + expect(logger.output).to match(/ERROR -- : \[LDClient\] Unable to load flag data from "#{Regexp.escape(file.path)}": /) + expect(status.state).to eq Interfaces::DataSource::Status::INITIALIZING + expect(status.last_error.kind).to eq Interfaces::DataSource::ErrorInfo::INVALID_DATA + end + end + + it "treats a missing file as a failed load with the same message" do + missing = File.join(@tmp_dir, "no-such-file.json") + + with_data_source({ paths: [missing] }) do |ds| + ds.start + expect(ds.initialized?).to be false + expect(logger.output).to include("Unable to load flag data from \"#{missing}\"") + end + end + + it "reports a duplicate key with the data kind namespace" do + file1 = make_temp_file({ flags: { flag1: flag_json("flag1") } }.to_json) + file2 = make_temp_file({ flagValues: { flag1: "x" } }.to_json) + + with_data_source({ paths: [file1.path, file2.path] }) do |ds| + ds.start + expect(logger.output).to include('features key "flag1" was used more than once') + expect(@store.init_count).to eq 0 + end + end + + it "compares only the modification time when polling" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + with_data_source({ paths: [file.path], auto_update: true, force_polling: true, poll_interval: 0.1 }) do |ds| + ds.start + mtime = File.mtime(file.path) + IO.write(file, { flagValues: { value1: "a much longer value than before" } }.to_json) + File.utime(mtime, mtime, file.path) + sleep 0.5 + + expect(@store.init_count).to eq 1 + end + end + + it "does not react to a deleted file when polling" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + with_data_source({ paths: [file.path], auto_update: true, force_polling: true, poll_interval: 0.1 }) do |ds| + ds.start + file.close + File.delete(file.path) + sleep 0.5 + + expect(@store.init_count).to eq 1 + expect(status.state).to eq Interfaces::DataSource::Status::VALID + expect(@store.get(Impl::DataStore::FEATURES, "value1")).not_to be_nil + end + end + + it "reloads on every polling interval after the first change" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + backdate(file) + + with_data_source({ paths: [file.path], auto_update: true, force_polling: true, poll_interval: 0.1 }) do |ds| + ds.start + IO.write(file, { flagValues: { value1: "y" } }.to_json) + expect(wait_for { @store.init_count >= 2 }).to be true + + expect(wait_for { @store.init_count >= 4 }).to be true + end + end + + it "uses the listen gem for auto-update when it is available" do + skip "the listen gem is not installed" unless defined?(Listen) + + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + expect(Listen).to receive(:to).and_call_original + + with_data_source({ paths: [file.path], auto_update: true }) do |ds| + ds.start + end + end + + it "does not use the listen gem when polling is forced" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + expect(Listen).not_to receive(:to) if defined?(Listen) + + with_data_source({ paths: [file.path], auto_update: true, force_polling: true, poll_interval: 0.1 }) do |ds| + ds.start + expect(Thread.list.map(&:name)).to include("LD/FileDataSource") + end + end + end + + describe Impl::Integrations::FileDataSourceV2 do + def no_selector_store + store = Object.new + store.define_singleton_method(:selector) { Interfaces::DataSystem::Selector.no_selector } + store + end + + def fetch_changes(paths) + source = Impl::Integrations::FileDataSourceV2.new(logger, paths: paths) + begin + result = source.fetch(no_selector_store) + expect(result.success?).to be true + result.value.change_set.changes + ensure + source.stop + end + end + + def listen_enabled? + Impl::Integrations::FileDataSourceV2.class_variable_get(:@@have_listen) + end + + def without_listen + klass = Impl::Integrations::FileDataSourceV2 + had_listen = listen_enabled? + klass.class_variable_set(:@@have_listen, false) + begin + yield + ensure + klass.class_variable_set(:@@have_listen, had_listen) + end + end + + def poller_threads + Thread.list.select { |t| t.name == "LD/FileDataSourceV2" } + end + + def with_sync(paths, poll_interval: 0.1) + source = Impl::Integrations::FileDataSourceV2.new(logger, paths: paths, poll_interval: poll_interval) + updates = Queue.new + pollers_before = poller_threads + thread = Thread.new { source.sync(no_selector_store) { |update| updates << update } } + begin + initial = pop_with_timeout(updates, 5) + expect(initial).not_to be_nil + expect(initial.state).to eq Interfaces::DataSource::Status::VALID + # The source starts change detection after it yields the initial data. A polling + # source has recorded the files' modification times once its poller thread exists, + # so a change made after that is one the poller can see. + expect(wait_for { (poller_threads - pollers_before).any? }).to be true unless listen_enabled? + yield updates + ensure + source.stop + thread.join(2) + end + end + + it "defaults a missing version to 1 and keeps a given version" do + file = make_temp_file({ flags: { flag1: flag_json("flag1"), flag2: flag_json("flag2", version: 7) }, + flagValues: { value1: "x" }, segments: { seg1: { key: "seg1" } } }.to_json) + + versions = fetch_changes([file.path]).to_h { |change| [change.key, change.version] } + + expect(versions).to eq({ flag1: 1, flag2: 7, value1: 1, seg1: 1 }) + end + + it "expands a flag value into a flag that is on and serves the value through its fallthrough" do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + change = fetch_changes([file.path]).detect { |c| c.key == :value1 } + expect(change.object[:on]).to be true + expect(change.object[:fallthrough]).to eq({ variation: 0 }) + expect(change.object).not_to have_key(:offVariation) + expect(change.object[:variations]).to eq ["x"] + end + + it "stores an item under its own key member rather than the key it appears under" do + file = make_temp_file({ flags: { "map-key": flag_json("own-key") } }.to_json) + + expect(fetch_changes([file.path]).map(&:key)).to eq [:"own-key"] + end + + it "reports a failed load with the file path" do + file = make_temp_file('{"flagValues"') + source = Impl::Integrations::FileDataSourceV2.new(logger, paths: [file.path]) + begin + result = source.fetch(no_selector_store) + + expect(result.success?).to be false + expect(result.error).to start_with("Unable to load flag data from \"#{file.path}\": ") + expect(logger.output).to match(/ERROR -- : \[LDClient\] Unable to load flag data from "#{Regexp.escape(file.path)}": /) + ensure + source.stop + end + end + + it "reports a duplicate key with the section name" do + file1 = make_temp_file({ flags: { flag1: flag_json("flag1") } }.to_json) + file2 = make_temp_file({ flagValues: { flag1: "x" } }.to_json) + source = Impl::Integrations::FileDataSourceV2.new(logger, paths: [file1.path, file2.path]) + begin + result = source.fetch(no_selector_store) + + expect(result.success?).to be false + expect(result.error).to include('In flags, key "flag1" was used more than once') + ensure + source.stop + end + end + + it "uses the listen gem for change detection when it is available" do + skip "the listen gem is not installed" unless defined?(Listen) + + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + listened = Queue.new + allow(Listen).to receive(:to).and_wrap_original do |original, *args, **options, &block| + listened << true + original.call(*args, **options, &block) + end + + with_sync([file.path]) do |_updates| + # The source starts the listener after it yields the initial data. + expect(pop_with_timeout(listened, 5)).to be true + end + end + + it "compares only the modification time when polling" do + without_listen do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + with_sync([file.path]) do |updates| + mtime = File.mtime(file.path) + IO.write(file, { flagValues: { value1: "a much longer value than before" } }.to_json) + File.utime(mtime, mtime, file.path) + + expect(pop_with_timeout(updates, 0.6)).to be_nil + end + end + end + + it "does not react to a deleted file when polling" do + without_listen do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + + with_sync([file.path]) do |updates| + file.close + File.delete(file.path) + + expect(pop_with_timeout(updates, 0.6)).to be_nil + end + end + end + + it "reloads once per change when polling" do + without_listen do + file = make_temp_file({ flagValues: { value1: "x" } }.to_json) + backdate(file) + + with_sync([file.path]) do |updates| + IO.write(file, { flagValues: { value1: "y" } }.to_json) + + update = pop_with_timeout(updates, 5) + expect(update).not_to be_nil + expect(update.state).to eq Interfaces::DataSource::Status::VALID + expect(pop_with_timeout(updates, 0.6)).to be_nil + end + end + end + end + end + end +end