Make the indexing schema version optional and fix review defects

The pipeline now treats `schema_version` as optional, so an ingestion format
with no versions (such as protobuf) can omit it. Each ingestion adapter decides
what a missing version means. The JSON adapter uses the latest available JSON
schema version, and still validates the event against that version, so a
malformed event still fails.

This removes the breaking changes the previous revision introduced:

- `Converters.upsert_event_for` accepts `__schema_version`, the legacy
  `__json_schema_version`, or neither. An existing generated project keeps
  working with no edit to `shared_factories.rb`.
- The JSON adapter claims and reads the legacy `json_schema_version` envelope
  key, so a direct caller of `Indexer#processor.process` needs no edit.
- The latency log and the warehouse dump log emit `schema_version` and also the
  deprecated alias `json_schema_version`, so existing dashboards keep working.
- The version selection log keeps its JSON-specific field names to match its
  JSON-specific message type `ElasticGraphMissingJSONSchemaVersion`.

It also fixes four defects found in review:

- `elasticgraph-indexer/README.md` and `indexing_event_decoder.rb` promised a
  default to the latest version that the code no longer had. Both texts are now
  format neutral, and the JSON gem documents its own behaviour.
- A duplicate "Indexing Event Decoder" section in the JSON gem README named a
  class that does not exist. Removed.
- `IndexerExtension#ingestion_adapters` memoized into the shared name
  `@ingestion_adapters` while calling `super`, which assigned the same name.
- `select_schema_version` sorted the available versions on every event.

The warehouse dumper uses the fixed S3 key segment `unversioned` in place of
`v<version>` for a version-less format, so the segment count stays the same.

Author: jwils

elasticgraph-indexer/README.md

@@ -90,6 +90,7 @@ module MyCompany
 end
 ```
 
-Decoded event hashes do not need to provide a schema version. When a version is omitted, the latest
-available schema artifact version is used for validation and record preparation. Decoders may include
-`schema_version` to request a specific schema artifact version.
+A decoded event hash may carry a `schema_version` to request a specific schema artifact version. The
+key is optional, because an ingestion format may have no versions at all. Each ingestion adapter
+decides what a missing version means for its own format. `elasticgraph-json_ingestion` uses the latest
+available JSON schema version.

elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb

@@ -22,8 +22,9 @@ def initialize(config:, schema_artifacts:, logger:)
         end
 
         # @param payload [String] a raw payload from the transport
-        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events. Events do not
-        #   need to include a schema version; when omitted, the latest available schema version is used.
+        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events. An event may
+        #   include a `schema_version`, but does not have to: an ingestion format with no versions
+        #   omits it. Each ingestion adapter decides what a missing version means for its own format.
         def decode(payload)
           # :nocov: -- must return an array to satisfy Steep type checking but never called
           []

elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb

@@ -36,6 +36,9 @@ def handles_event?(event)
         # Validates the given event and resolves the record preparer appropriate for the event's
         # schema version.
         #
+        # The event's `schema_version` is optional, because an ingestion format may have no versions
+        # at all. Each adapter decides what a missing version means for its own format.
+        #
         # @param event [Hash<String, Object>] an ElasticGraph indexing event
         # @return [ValidationResult] the result of validating the event
         def validate_event(event)

elasticgraph-indexer/lib/elastic_graph/indexer/processor.rb

@@ -120,12 +120,18 @@ def calculate_latency_metrics(successful_operations, noop_results)
 
           result = successful_events.include?(event) ? "success" : "noop"
 
+          # The schema version is optional, since an ingestion format may have no versions at all.
+          schema_version = event[SCHEMA_VERSION_KEY]
+
           @logger.info({
             "message_type" => "ElasticGraphIndexingLatencies",
             "message_id" => event["message_id"],
             "event_type" => event.fetch("type"),
             "event_id" => EventID.from_event(event).to_s,
-            SCHEMA_VERSION_KEY => event.fetch(SCHEMA_VERSION_KEY),
+            SCHEMA_VERSION_KEY => schema_version,
+            # Deprecated alias of `schema_version`, kept so that dashboards and monitors that watch
+            # the old name keep working.
+            JSON_SCHEMA_VERSION_KEY => schema_version,
             "latencies_in_ms_from" => latencies_in_ms_from,
             "slo_results" => slo_results,
             "result" => result

elasticgraph-indexer/lib/elastic_graph/indexer/test_support/converters.rb

@@ -14,16 +14,27 @@ module ElasticGraph
   class Indexer
     module TestSupport
       module Converters
+        # Attributes that describe the event rather than the record, so they never reach the record.
+        # `__json_schema_version` is the legacy name of `__schema_version`; projects generated before
+        # the schema version became ingestion-format-neutral still use it.
+        EVENT_ONLY_ATTRIBUTES = ["__typename", "__version", "__schema_version", "__json_schema_version"]
+
         # Helper method for testing and generating fake data to convert a factory record into an event
         def self.upsert_event_for(record)
-          {
+          event = {
             "op" => "upsert",
             "id" => record.fetch("id"),
             "type" => record.fetch("__typename"),
             "version" => record.fetch("__version"),
-            "record" => record.except("__typename", "__version", "__schema_version"),
-            SCHEMA_VERSION_KEY => record.fetch("__schema_version")
+            "record" => record.except(*EVENT_ONLY_ATTRIBUTES)
           }
+
+          # The schema version is optional, so include it only when the factory supplies one.
+          if (schema_version = record["__schema_version"] || record["__json_schema_version"])
+            event[SCHEMA_VERSION_KEY] = schema_version
+          end
+
+          event
         end
 
         # Helper method to create an array of events given an array of records

elasticgraph-indexer/sig/elastic_graph/indexer/test_support/converters.rbs

@@ -2,6 +2,8 @@ module ElasticGraph
   class Indexer
     module TestSupport
       module Converters
+        EVENT_ONLY_ATTRIBUTES: ::Array[::String]
+
         def self.upsert_event_for: (::Hash[::String, untyped]) -> ::Hash[::String, untyped]
 
         def self.upsert_events_for_records: (

elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb

@@ -127,10 +127,10 @@ module Operation
             expect_failed_event_error(event, "missing_keys", "type", expect_no_ops: true)
           end
 
-          it "notifies an error on missing `#{SCHEMA_VERSION_KEY}`" do
+          it "builds operations for an event that carries no `#{SCHEMA_VERSION_KEY}`, since the key is optional" do
             event = build_upsert_event(:component).except(SCHEMA_VERSION_KEY)
 
-            expect_failed_event_error(event, SCHEMA_VERSION_KEY)
+            expect(build_expecting_success(event)).not_to be_empty
           end
 
           it "notifies an error on wrong field types" do

elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb

@@ -135,7 +135,20 @@ class Indexer
 
             expect(logged_jsons_of_type("ElasticGraphIndexingLatencies").first).to include(
               "event_id" => "Component:#{component.fetch("id")}@v#{component.fetch("version")}",
-              "message_id" => "m1"
+              "message_id" => "m1",
+              SCHEMA_VERSION_KEY => component.fetch(SCHEMA_VERSION_KEY),
+              # Deprecated alias, kept for existing dashboards and monitors.
+              JSON_SCHEMA_VERSION_KEY => component.fetch(SCHEMA_VERSION_KEY)
+            )
+          end
+
+          it "logs a nil schema version for an event that carries none, since the key is optional" do
+            component = upsert_event_with_latency_timestamps(:component, 36, 72).except(SCHEMA_VERSION_KEY)
+            process([component])
+
+            expect(logged_jsons_of_type("ElasticGraphIndexingLatencies").first).to include(
+              SCHEMA_VERSION_KEY => nil,
+              JSON_SCHEMA_VERSION_KEY => nil
             )
           end
 

elasticgraph-indexer/spec/unit/elastic_graph/indexer/test_support/converters_spec.rb

@@ -33,6 +33,42 @@ module TestSupport
               SCHEMA_VERSION_KEY => 1
             )
           end
+
+          it "accepts the legacy `__json_schema_version` attribute, so factories from older projects keep working" do
+            factory_record = {
+              "id" => "1",
+              "__version" => 1,
+              "__typename" => "Widget",
+              "__json_schema_version" => 3,
+              "field1" => "value1"
+            }
+
+            expect(TestSupport::Converters.upsert_event_for(factory_record)).to eq(
+              "op" => "upsert",
+              "id" => "1",
+              "version" => 1,
+              "type" => "Widget",
+              "record" => {"id" => "1", "field1" => "value1"},
+              SCHEMA_VERSION_KEY => 3
+            )
+          end
+
+          it "omits the schema version when the factory record supplies none, since the key is optional" do
+            factory_record = {
+              "id" => "1",
+              "__version" => 1,
+              "__typename" => "Widget",
+              "field1" => "value1"
+            }
+
+            expect(TestSupport::Converters.upsert_event_for(factory_record)).to eq(
+              "op" => "upsert",
+              "id" => "1",
+              "version" => 1,
+              "type" => "Widget",
+              "record" => {"id" => "1", "field1" => "value1"}
+            )
+          end
         end
 
         describe ".upsert_events_for_records" do

elasticgraph-json_ingestion/README.md

@@ -7,19 +7,6 @@ events and validates JSON-ingestion-specific schema options. Generated ElasticGr
 and enable it by default. Applications that wire schema-definition tasks manually enable it by adding
 `ElasticGraph::JSONIngestion::SchemaDefinition::APIExtension` to their schema-definition extension modules.
 
-## Indexing Event Decoder
-
-JSON ingestion payloads can include `json_schema_version` to request a specific JSON schema artifact
-version. Configure the JSON ingestion decoder when the indexer consumes those payloads so the JSON-specific
-field is mapped to the indexer's generic `schema_version` event field.
-
-```yaml
-indexer:
-  indexing_event_decoder:
-    name: ElasticGraph::JSONIngestion::IndexingEventDecoder::JSONLines
-    require_path: elastic_graph/json_ingestion/indexing_event_decoder
-```
-
 ## Schema Definition APIs
 
 Use `schema.json_schema_version` to identify the current JSON schema artifact. Every change that affects
@@ -119,11 +106,22 @@ end
 
 Beyond schema definition, this gem teaches `elasticgraph-indexer` how to ingest JSON events: it provides an
 ingestion adapter that validates each event against the JSON schema identified by the event's
-`json_schema_version` and prepares its record for indexing using that version's view of the schema.
+`schema_version` and prepares its record for indexing using that version's view of the schema.
 
 No configuration is needed: defining your schema with this gem's `SchemaDefinition::APIExtension` registers
 an indexer extension in your schema artifacts' runtime metadata, which the indexer applies when it boots.
 
+### Schema versions
+
+The adapter resolves the version of each event as follows:
+
+- The `schema_version` key selects the JSON schema version. When the exact version is unavailable, the
+  adapter selects the closest available version and logs `ElasticGraphMissingJSONSchemaVersion`.
+- The legacy `json_schema_version` key still works, so a publisher or an in-process caller that predates
+  the ingestion-format-neutral key needs no change.
+- An event that carries neither key gets the latest available JSON schema version. The adapter still
+  validates the event against that version, so a malformed event still fails.
+
 This gem also provides the `be_a_valid_elastic_graph_event` RSpec matcher (via
 `require "elastic_graph/json_ingestion/spec_support/event_matcher"`) for testing that publisher events
 conform to your schema.