From eaf2a9cf2d4d22d5c1cc6c15bb59355fd1ba43bf Mon Sep 17 00:00:00 2001 From: Ashit Verma Date: Tue, 21 Jul 2026 13:42:37 +0100 Subject: [PATCH] Add `skip_record_validation_for` indexer config for sampled backfill validation Adds an indexer config option that skips per-record JSON schema validation for a configurable fraction of records, keyed by GraphQL type. It exists for backfills of trusted, pre-validated data, where the per-record schema walk is a meaningful ingest cost and the datastore mappings provide a coarse backstop. `skip_record_validation_for` maps a type name to a fraction in `[0.0, 1.0]`: `0.0` (or an absent key) validates every record, `1.0` skips every record, and values in between sample. The skip decision is deterministic per event id -- a stable `Zlib.crc32` of `EventID#to_s` buckets each event -- so the same event makes the same decision on retry across indexer pods (`String#hash` is unsuitable: `RUBY_HASH_SEED` is per-process). The event envelope is always validated regardless of the sampling rate. Two supporting pieces: - `RecordPreparer::UnknownTypeError` (a `KeyError` subtype) is raised when a skipped record reaches the preparer with a missing/unknown abstract-type `__typename`. `Factory#build` rescues it and returns a structured `FailedEventError` rather than letting an exception escape, with a message that omits the offending value. - Skipping a safety check is never silent: `Processor` tallies skipped records per batch and logs a single aggregate `RecordValidationSkipped` entry (with per-type counts), mirroring the batch-level `ElasticGraphIndexingLatencies` log. Per-record logging would be untenable at backfill scale. --- .../lib/elastic_graph/indexer.rb | 1 + .../lib/elastic_graph/indexer/config.rb | 25 ++- .../indexer/operation/factory.rb | 53 ++++-- .../lib/elastic_graph/indexer/processor.rb | 17 ++ .../elastic_graph/indexer/record_preparer.rb | 12 +- .../sig/elastic_graph/indexer/config.rbs | 4 + .../indexer/operation/factory.rbs | 14 +- .../sig/elastic_graph/indexer/processor.rbs | 1 + .../elastic_graph/indexer/record_preparer.rbs | 3 + .../unit/elastic_graph/indexer/config_spec.rb | 27 +++ .../indexer/operation/factory_spec.rb | 172 ++++++++++++++++++ .../elastic_graph/indexer/processor_spec.rb | 41 +++++ .../local/spec_support/config_schema.yaml | 25 +++ 13 files changed, 377 insertions(+), 18 deletions(-) diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index 218b6818d..b1b47c3fc 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -85,6 +85,7 @@ def operation_factory record_preparer_factory: record_preparer_factory, logger: datastore_core.logger, skip_derived_indexing_type_updates: config.skip_derived_indexing_type_updates, + skip_record_validation_for: config.skip_record_validation_for, configure_record_validator: nil ) end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb index 6eca6a902..b5de165b7 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb @@ -11,7 +11,7 @@ module ElasticGraph class Indexer - class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :extension_modules) + class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :skip_record_validation_for, :extension_modules) json_schema at: "indexer", optional: false, description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.", @@ -43,15 +43,36 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms {"WidgetWorkspace" => ["ABC12345678"]} ] }, + skip_record_validation_for: { + description: "Map of GraphQL type names to the fraction of records (in `[0.0, 1.0]`) whose " \ + "per-type JSON schema validation should be skipped. `0.0` (or an absent key) validates every record; " \ + "`1.0` skips every record; values in between sample. The decision is deterministic per event id " \ + "(`type:id@vversion`) so the same event makes the same skip decision on retry across indexer pods. " \ + "The event envelope (op, id, type, version, json_schema_version, latency_timestamps) is always " \ + "validated. Intended for backfills of trusted, pre-validated data where skipping the per-record " \ + "schema walk yields meaningful ingest speedups, with a sampled fraction left validated as a canary " \ + "for schema drift. Leave empty for live-traffic ingestion: the datastore mappings will not catch all " \ + "the constraints (regex, enum, min/max, format, abstract-type discriminators) that the JSON schema " \ + "enforces.", + type: "object", + patternProperties: {/^[A-Z]\w*$/.source => {type: "number", minimum: 0, maximum: 1}}, + additionalProperties: false, + default: {}, # : untyped + examples: [ + {}, # : untyped + {"Widget" => 0.9, "Component" => 1.0} + ] + }, extension_modules: Support::Config::EXTENSION_MODULE_SCHEMA } private - def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, extension_modules:) + def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, skip_record_validation_for:, extension_modules:) { skip_derived_indexing_type_updates: skip_derived_indexing_type_updates.transform_values(&:to_set), latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms, + skip_record_validation_for: skip_record_validation_for.transform_values(&:to_f), extension_modules: SchemaArtifacts::RuntimeMetadata::ExtensionLoader.load_component_extensions(extension_modules) } end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb index c6b5365de..aee833913 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb @@ -13,6 +13,7 @@ require "elastic_graph/indexer/record_preparer" require "elastic_graph/support/json_schema/validator_factory" require "elastic_graph/support/memoizable_data" +require "zlib" module ElasticGraph class Indexer @@ -23,6 +24,7 @@ class Factory < Support::MemoizableData.define( :record_preparer_factory, :logger, :skip_derived_indexing_type_updates, + :skip_record_validation_for, :configure_record_validator ) def build(event) @@ -40,11 +42,27 @@ def build(event) return build_failed_result(event, "event payload", error_message) end - failed_result = validate_record_returning_failure(event, selected_json_schema_version) - failed_result || BuildResult.success(build_all_operations_for( - event, - record_preparer_factory.for_json_schema_version(selected_json_schema_version) - )) + graphql_type_name = event.fetch("type") + validation_skipped = skip_validation?(graphql_type_name, event) + + unless validation_skipped + failed_result = validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version) + return failed_result if failed_result + end + + begin + BuildResult.success( + build_all_operations_for(event, record_preparer_factory.for_json_schema_version(selected_json_schema_version)), + validation_skipped_for: validation_skipped ? graphql_type_name : nil + ) + rescue RecordPreparer::UnknownTypeError + # Safety net for `skip_record_validation_for`: when record validation is skipped, an + # event with a missing or unknown abstract-type `__typename` reaches `RecordPreparer` + # and raises. Convert it to a `FailedEventError` so callers see a structured failure + # rather than an exception leaking out of `Factory#build`. The message intentionally + # omits the offending value to avoid leaking record data. + build_failed_result(event, "#{graphql_type_name} record", "Missing or unknown `__typename` for an abstract-type field.") + end end private @@ -117,9 +135,8 @@ def prepare_event(event) event.merge("record" => event["record"].merge("id" => event.fetch("id"))) end - def validate_record_returning_failure(event, selected_json_schema_version) + def validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version) record = event.fetch("record") - graphql_type_name = event.fetch("type") validator = validator(graphql_type_name, selected_json_schema_version) if (error_message = validator.validate_with_error_message(record)) @@ -127,6 +144,18 @@ def validate_record_returning_failure(event, selected_json_schema_version) end end + # Decides whether to skip per-record validation for `event` of `type`. The decision is + # deterministic per event id: a stable `Zlib.crc32` of `EventID#to_s` puts each event in a + # bucket in `[0.0, 1.0)` that is compared against the configured skip rate. Same event id => + # same decision across pods and retries, so retries do not flip records between + # validated/skipped. `String#hash` is unsuitable here: `RUBY_HASH_SEED` is per-process. + def skip_validation?(type, event) + rate = skip_record_validation_for[type] + return false if rate.nil? || rate <= 0.0 + return true if rate >= 1.0 + ::Zlib.crc32(EventID.from_event(event).to_s).fdiv(2**32) < rate + end + def build_failed_result(event, payload_description, validation_message) message = "Malformed #{payload_description}. #{validation_message}" @@ -192,14 +221,16 @@ def raise(*args) # Return value from `build` that indicates what happened. # - If it was successful, `operations` will be a non-empty array of operations and `failed_event_error` will be nil. # - If there was a validation issue, `operations` will be an empty array and `failed_event_error` will be non-nil. - BuildResult = ::Data.define(:operations, :failed_event_error) do + # - `validation_skipped_for` names the event's GraphQL type when per-record validation was skipped + # (via `skip_record_validation_for`), and is nil otherwise. `Processor` aggregates this for observability. + BuildResult = ::Data.define(:operations, :failed_event_error, :validation_skipped_for) do # @implements BuildResult - def self.success(operations) - new(operations, nil) + def self.success(operations, validation_skipped_for: nil) + new(operations, nil, validation_skipped_for) end def self.failure(failed_event_error) - new([], failed_event_error) + new([], failed_event_error, nil) end end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb index 4f8064c39..3e87675bc 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb @@ -45,6 +45,8 @@ def process_returning_failures(events, refresh_indices: false) factory_results = factory_results_by_event.values + log_skipped_record_validations(factory_results) + bulk_result = @datastore_router.bulk(factory_results.flat_map(&:operations), refresh: refresh_indices) successful_operations = bulk_result.successful_operations(check_failures: false) @@ -62,6 +64,21 @@ def process_returning_failures(events, refresh_indices: false) private + # Emits a single aggregate log line per batch when any records had their per-record validation + # skipped (via `skip_record_validation_for`). Skipping a safety check should never be silent, but + # per-record logging would be untenable at backfill scale (a `1.0` skip rate is one line per record), + # so we tally by type and log once. Mirrors the batch-level `ElasticGraphIndexingLatencies` log. + def log_skipped_record_validations(factory_results) + counts_by_type = factory_results.filter_map(&:validation_skipped_for).tally + return if counts_by_type.empty? + + @logger.info({ + "message_type" => "RecordValidationSkipped", + "count" => counts_by_type.values.sum, + "counts_by_type" => counts_by_type + }) + end + def categorize_failures(failures, events) source_event_versions_by_cluster_by_op = @datastore_router.source_event_versions_in_index( failures.flat_map { |f| f.versioned_operations.to_a } diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb index 6e0b04e44..c92ec8d4a 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb @@ -11,6 +11,13 @@ module ElasticGraph class Indexer class RecordPreparer + # Raised when an abstract-type record is missing a `__typename` (or carries one that does not + # name a known concrete subtype). Normally JSON-schema validation prevents reaching here, but + # `Indexer::Config#skip_record_validation_for` lets that validation be skipped -- so this error + # exists as a typed, catchable signal that callers can convert into a structured failure. + class UnknownTypeError < ::KeyError + end + # Provides the ability to get a `RecordPreparer` for a specific JSON schema version. class Factory def initialize(schema_artifacts) @@ -116,7 +123,10 @@ def prepare_for_index(type_name, value, mapping_properties) # # If `type_name` is an abstract type, we need to look at the `__typename` field to see # what the concrete subtype is. `__typename` is required on abstract types and indicates that. - eg_meta_by_field_name = @eg_meta_by_field_name_by_concrete_type.fetch(value["__typename"] || type_name) + concrete_type = value["__typename"] || type_name + eg_meta_by_field_name = @eg_meta_by_field_name_by_concrete_type.fetch(concrete_type) do + raise UnknownTypeError, "No concrete type named `#{concrete_type}` is known to the record preparer." + end # We only want to consider __typename if it's in the per-record mapping in order to determine # whether __typename is required on records. When it's a constant_keyword it exists at the index diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs index a4f343d59..d06d49697 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs @@ -5,16 +5,19 @@ module ElasticGraph attr_reader latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer] attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] + attr_reader skip_record_validation_for: ::Hash[::String, ::Float] attr_reader extension_modules: ::Array[::Module] def initialize: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + ?skip_record_validation_for: ::Hash[::String, ::Float], ?extension_modules: ::Array[::Module]) -> void def with: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + ?skip_record_validation_for: ::Hash[::String, ::Float], ?extension_modules: ::Array[::Module]) -> Config def self.members: () -> ::Array[::Symbol] @@ -26,6 +29,7 @@ module ElasticGraph def convert_values: ( latency_slo_thresholds_by_timestamp_in_ms: untyped, skip_derived_indexing_type_updates: untyped, + skip_record_validation_for: untyped, extension_modules: untyped ) -> ::Hash[::Symbol, untyped] diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs index 82e968053..1f109376c 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs @@ -11,6 +11,7 @@ module ElasticGraph attr_reader record_preparer_factory: RecordPreparer::Factory attr_reader logger: ::Logger attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] + attr_reader skip_record_validation_for: ::Hash[::String, ::Float] attr_reader configure_record_validator: (^(validatorFactory) -> validatorFactory)? def initialize: ( @@ -19,6 +20,7 @@ module ElasticGraph record_preparer_factory: RecordPreparer::Factory, logger: ::Logger, skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + skip_record_validation_for: ::Hash[::String, ::Float], configure_record_validator: (^(validatorFactory) -> validatorFactory)? ) -> void @@ -28,6 +30,7 @@ module ElasticGraph ?record_preparer_factory: RecordPreparer::Factory, ?logger: ::Logger, ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + ?skip_record_validation_for: ::Hash[::String, ::Float], ?configure_record_validator: (^(validatorFactory) -> validatorFactory)? ) -> instance end @@ -44,7 +47,8 @@ module ElasticGraph def select_json_schema_version: (event) { (BuildResult) -> bot } -> (::Integer | bot) def prepare_event: (event) -> event - def validate_record_returning_failure: (event, ::Integer) -> BuildResult? + def validate_record_returning_failure: (event, ::String, ::Integer) -> BuildResult? + def skip_validation?: (::String, event) -> bool def build_failed_result: (event, ::String, ::String) -> BuildResult def build_all_operations_for: (event, _RecordPreparer) -> ::Array[_Operation] def index_definitions_for: (::String) -> ::Array[DatastoreCore::_IndexDefinition] @@ -53,15 +57,17 @@ module ElasticGraph class BuildResult attr_reader operations: ::Array[_Operation] attr_reader failed_event_error: FailedEventError? + attr_reader validation_skipped_for: ::String? - def initialize: (::Array[_Operation], FailedEventError?) -> void + def initialize: (::Array[_Operation], FailedEventError?, ::String?) -> void def with: ( ?operations: ::Array[_Operation], - ?failed_event_error: FailedEventError? + ?failed_event_error: FailedEventError?, + ?validation_skipped_for: ::String? ) -> BuildResult - def self.success: (::Array[_Operation]) -> BuildResult + def self.success: (::Array[_Operation], ?validation_skipped_for: ::String?) -> BuildResult def self.failure: (FailedEventError) -> BuildResult end end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs index ac450f43a..0f34056ce 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs @@ -20,6 +20,7 @@ module ElasticGraph @indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer] @clock: singleton(::Time) + def log_skipped_record_validations: (::Array[Operation::Factory::BuildResult]) -> void def categorize_failures: (::Array[FailedEventError], ::Array[event]) -> ::Array[FailedEventError] def calculate_latency_metrics: (::Array[_Operation], ::Array[Operation::Result]) -> void end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs index 1c4efcefc..6401aa8f7 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs @@ -8,6 +8,9 @@ module ElasticGraph type egMetaByFieldHash = ::Hash[::String, {"type" => ::String, "nameInIndex" => ::String}] type egMetaByFieldByTypeHash = ::Hash[::String, egMetaByFieldHash] + class UnknownTypeError < ::KeyError[untyped, untyped] + end + class Factory def initialize: (schemaArtifacts) -> void diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb index 2c157a8e9..15a51d2e5 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb @@ -32,6 +32,33 @@ class Indexer expect(config.skip_derived_indexing_type_updates).to eq("WidgetCurrency" => ["USD"].to_set) end + it "coerces `skip_record_validation_for` rates to floats (so integer YAML values like `1` become `1.0`)" do + config = Config.from_parsed_yaml("indexer" => { + "latency_slo_thresholds_by_timestamp_in_ms" => {}, + "skip_record_validation_for" => { + "Widget" => 1, + "Component" => 0.5 + } + }) + + expect(config.skip_record_validation_for).to eq("Widget" => 1.0, "Component" => 0.5) + expect(config.skip_record_validation_for.values).to all(be_a(::Float)) + end + + it "rejects `skip_record_validation_for` rates outside `[0.0, 1.0]`" do + expect { + Config.from_parsed_yaml("indexer" => { + "skip_record_validation_for" => {"Widget" => 1.5} + }) + }.to raise_error Errors::ConfigError + + expect { + Config.from_parsed_yaml("indexer" => { + "skip_record_validation_for" => {"Widget" => -0.1} + }) + }.to raise_error Errors::ConfigError + end + describe "#extension_modules", :in_temp_dir do it "loads the extension modules from disk" do File.write("eg_extension_module1.rb", <<~EOS) diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb index 84a85ddc0..12697d70d 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb @@ -75,6 +75,178 @@ module Operation end end + # We deliberately construct the indexer here without going through `build_indexer`. The + # `skip_record_validation_for` knob is intentionally not exposed via spec helpers so that + # tests cannot silently weaken validation; enabling it must be a visible, deliberate choice + # in each spec that exercises it. + context "when the indexer is configured to skip record validation for some types" do + let(:indexer) do + datastore_core = build_datastore_core + Indexer.new( + datastore_core: datastore_core, + config: Indexer::Config.new( + latency_slo_thresholds_by_timestamp_in_ms: {}, + skip_derived_indexing_type_updates: {}, + skip_record_validation_for: {"Component" => 1.0} + ) + ) + end + + it "skips per-type record validation for the listed type but still builds operations" do + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 # would normally fail JSON schema validation + + expect(build_expecting_success(event)).to eq([new_primary_indexing_operation({ + "op" => "upsert", + "id" => "1", + "type" => "Component", + "version" => 1, + "record" => event["record"], + JSON_SCHEMA_VERSION_KEY => 1 + })]) + end + + it "records the skipped type on the successful build result" do + event = build_upsert_event(:component, id: "1", __version: 1) + + result = indexer.operation_factory.build(event) + + expect(result.validation_skipped_for).to eq("Component") + end + + it "still validates record-level fields for types that are not in the skip list" do + widget_event = build_upsert_event(:widget, id: "1", __version: 1) + widget_event["record"]["name"] = 123 + + expect_failed_event_error(widget_event, "Malformed Widget record", "name") + end + + it "does not flag validated types on the successful build result" do + event = build_upsert_event(:widget, id: "1", __version: 1) + + result = indexer.operation_factory.build(event) + + expect(result.validation_skipped_for).to be_nil + end + + it "still applies envelope-level validation for skipped types" do + event = build_upsert_event(:component, id: "1", __version: -1) + + expect_failed_event_error(event, "/properties/version") + end + end + + context "when the indexer configures a fractional skip rate for a type" do + let(:indexer) do + datastore_core = build_datastore_core + Indexer.new( + datastore_core: datastore_core, + config: Indexer::Config.new( + latency_slo_thresholds_by_timestamp_in_ms: {}, + skip_derived_indexing_type_updates: {}, + skip_record_validation_for: {"Component" => 0.5} + ) + ) + end + + it "skips validation when the event's stable bucket falls below the rate" do + # Stub crc32 so the bucket is 0.0 -- well below 0.5 -- forcing the "skip" branch. + allow(::Zlib).to receive(:crc32).and_return(0) + + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 # would normally fail JSON schema validation + + expect { + build_expecting_success(event) + }.not_to raise_error + end + + it "still validates when the event's stable bucket falls at or above the rate" do + # Stub crc32 to a value just above 0.5 * 2**32 so the bucket >= rate, forcing validation. + allow(::Zlib).to receive(:crc32).and_return((2**32 * 0.75).to_i) + + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 + + expect_failed_event_error(event, "Malformed Component record", "name") + end + + it "produces the same skip decision for the same event on retry" do + # No stubbing -- this exercises the real crc32 and locks in determinism without + # coupling to a specific hash output. Two builds of the same event must agree on + # whether to skip validation. + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 # would fail validation if not skipped + + first = indexer.operation_factory.build(event) + second = indexer.operation_factory.build(event) + + expect(first.failed_event_error.nil?).to eq(second.failed_event_error.nil?) + expect(first.operations.size).to eq(second.operations.size) + expect(first.validation_skipped_for).to eq(second.validation_skipped_for) + end + end + + context "when record validation is skipped for a type that has derived-index update targets" do + # Widget has a `WidgetCurrency` derived index update target. With validation skipped, + # `build_all_operations_for` still has to traverse `Update.operations_for` and the + # schema artifacts. This spec locks in that skipping does not regress the derived path. + let(:indexer) do + datastore_core = build_datastore_core + Indexer.new( + datastore_core: datastore_core, + config: Indexer::Config.new( + latency_slo_thresholds_by_timestamp_in_ms: {}, + skip_derived_indexing_type_updates: {}, + skip_record_validation_for: {"Widget" => 1.0} + ) + ) + end + + it "still emits both the primary and the derived-index update operations" do + event = build_upsert_event(:widget, id: "1", __version: 1) + formatted_event = { + "op" => "upsert", + "id" => "1", + "type" => "Widget", + "version" => 1, + "record" => event["record"], + JSON_SCHEMA_VERSION_KEY => 1 + } + + expect(build_expecting_success(event)).to contain_exactly( + new_primary_indexing_operation(formatted_event, index_def: index_def_named("widgets")), + widget_currency_derived_update_operation_for(formatted_event) + ) + end + end + + context "when validation is skipped and an abstract-type field carries an unknown `__typename`" do + # Widget has an `inventor` field whose type is the `Inventor` union. With validation + # skipped, `RecordPreparer` would normally raise on an unknown concrete subtype; the + # safety net in `Factory#build` converts that into a structured `FailedEventError`. + let(:indexer) do + datastore_core = build_datastore_core + Indexer.new( + datastore_core: datastore_core, + config: Indexer::Config.new( + latency_slo_thresholds_by_timestamp_in_ms: {}, + skip_derived_indexing_type_updates: {}, + skip_record_validation_for: {"Widget" => 1.0} + ) + ) + end + + it "produces a FailedEventError instead of raising" do + event = build_upsert_event(:widget, id: "1", __version: 1) + event["record"]["inventor"] = {"__typename" => "NotARealConcreteType", "name" => "anon"} + + expect { + expect_failed_event_error(event, "Widget record", "__typename") + }.not_to raise_error + end + end + it "generates a primary indexing operation for a single index with latency metrics" do event = build_upsert_event(:component, id: "1", __version: 1) latency_timestamps = {"latency_timestamps" => {"created_in_esperanto_at" => "2012-04-23T18:25:43.511Z"}} diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb index 79a47d7da..fbf380fdd 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb @@ -386,6 +386,47 @@ def make_component_bad(component) end end + context "when the indexer skips per-record validation for some types" do + # `skip_record_validation_for` is intentionally not exposed via `build_indexer`, so we + # construct the indexer directly (reusing the spec's router spy and clock) to enable it. + let(:indexer) do + Indexer.new( + datastore_core: build_datastore_core, + config: Indexer::Config.new( + latency_slo_thresholds_by_timestamp_in_ms: {}, + skip_derived_indexing_type_updates: {}, + skip_record_validation_for: {"Component" => 1.0} + ), + datastore_router: datastore_router, + clock: clock + ) + end + + it "logs a single aggregate `RecordValidationSkipped` entry per batch, counted by type" do + component1 = build_upsert_event(:component, id: "c1", __version: 1) + component2 = build_upsert_event(:component, id: "c2", __version: 1) + address = build_upsert_event(:address, id: "a1", __version: 1) + + process([component1, component2, address]) + + expect(logged_jsons_of_type("RecordValidationSkipped")).to contain_exactly( + a_hash_including( + "message_type" => "RecordValidationSkipped", + "count" => 2, + "counts_by_type" => {"Component" => 2} + ) + ) + end + + it "logs nothing when no records in the batch had validation skipped" do + address = build_upsert_event(:address, id: "a1", __version: 1) + + process([address]) + + expect(logged_jsons_of_type("RecordValidationSkipped")).to be_empty + end + end + def build_indexer_with(latency_thresholds:) build_indexer( clock: clock, diff --git a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml index 638a74a12..f94406a95 100644 --- a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml +++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml @@ -508,6 +508,31 @@ properties: - {} - WidgetWorkspace: - ABC12345678 + skip_record_validation_for: + description: 'Map of GraphQL type names to the fraction of records (in `[0.0, + 1.0]`) whose per-type JSON schema validation should be skipped. `0.0` (or + an absent key) validates every record; `1.0` skips every record; values + in between sample. The decision is deterministic per event id (`type:id@vversion`) + so the same event makes the same skip decision on retry across indexer pods. + The event envelope (op, id, type, version, json_schema_version, latency_timestamps) + is always validated. Intended for backfills of trusted, pre-validated data + where skipping the per-record schema walk yields meaningful ingest speedups, + with a sampled fraction left validated as a canary for schema drift. Leave + empty for live-traffic ingestion: the datastore mappings will not catch + all the constraints (regex, enum, min/max, format, abstract-type discriminators) + that the JSON schema enforces.' + type: object + patternProperties: + "^[A-Z]\\w*$": + type: number + minimum: 0 + maximum: 1 + additionalProperties: false + default: {} + examples: + - {} + - Widget: 0.9 + Component: 1.0 extension_modules: description: Array of modules that will be extended onto the component instance to support extension libraries.