Route events through a typed Event value by myronmarston · Pull Request #1398 · block/elasticgraph

myronmarston · 2026-09-18T03:48:23Z

Why

Every consumer of an indexing event reached into a Hash with string keys. The shape of the envelope lived only in scattered fetch calls, Steep could not check it, and each consumer had to repeat the defaults for optional fields. A Data value with typed members states the envelope once.

An Event exists only after an ingestion adapter validates its envelope, so every instance has the typed fields the envelope declares.

What

Indexer::Event, with members op, type, id, version, record, schema_version, ingestion_format, message_id, and latency_timestamps.
Typed accessors in place of event["x"] and event.fetch("x") across EventID, FailedEventError, IndexingFailuresError, Operation::Factory, Operation::Update, DatastoreIndexingRouter, Processor, IngestionAdapter, the JSON ingestion adapter, WarehouseDumper, and TestSupport::Converters.
FailedEventError delegates id, op, type, version, record, and message_id to its event through Forwardable, and includes IndexingFailuresError::_IndexingFailure so that Steep enforces the interface it satisfies.
The JSON adapter's events_from returns Events through Event.from_validated_hash.

Event#record keeps the native record type of the adapter, which is a Hash for JSON. An ingestion format with a different record type can therefore carry its own.

🤖 Generated with Claude Code

myronmarston · 2026-09-18T03:48:25Z

This change is part of the following stack:

Point test datastore URLs at 127.0.0.1 instead of localhost #1400
- Move record id preparation into the JSON ingestion adapter #1393
  - Report indexing failures through a _IndexingFailure interface #1395
    - Make _RecordPreparer generic #1396
      - Validate the event envelope before building operations #1397
        
        Route events through a typed Event value #1398 ◀

_{Change managed by git-spice.}

jwils · 2026-09-21T00:05:56Z

        },
        "nested_fields" => {
-          "max_widget_cost" => ([w1] + widgets).select { |w| w.dig("record", "cost", "currency") == "USD" }.map { |w| w.fetch("record").fetch("cost").fetch("amount_cents") }.max
+          "max_widget_cost" => ([w1] + widgets).select { |w| w.record.dig("cost", "currency") == "USD" }.map { |w| w.record.fetch("cost").fetch("amount_cents") }.max


Suggested change

"max_widget_cost" => ([w1] + widgets).select { |w| w.record.dig("cost", "currency") == "USD" }.map { |w| w.record.fetch("cost").fetch("amount_cents") }.max

"max_widget_cost" => ([w1] + widgets).filter_map do |w|

w.record.fetch("cost").fetch("amount_cents") if w.record.dig("cost", "currency") == "USD"

end.max

jwils · 2026-09-21T00:06:35Z

+      end
+    end
+
+    # Steep weirdly expects them here...


Do you know why?

jwils · 2026-09-21T00:16:30Z

+      # @param hash [Hash<String, Object>] a validated JSON event
+      # @return [Event]
+      def self.from_validated_hash(hash)
+        latency_timestamps = hash["latency_timestamps"] || {} # : ::Hash[::String, ::String]


Any reason not the follow the same pattern as the other entries?

Suggested change

latency_timestamps = hash["latency_timestamps"] || {} # : ::Hash[::String, ::String]

hash.fetch("latency_timestamps", {})

jwils · 2026-09-21T00:19:04Z

+          schema_version: hash.fetch(JSON_SCHEMA_VERSION_KEY),
+          ingestion_format: hash.fetch(INGESTION_FORMAT_KEY, "json"),


Is there a way to put this in the json_ingestion? I don't love having the json specific keys stay here.

- Introduce Indexer::Event, a Data value object with a validated envelope (op, type, id, version, record, schema_version, ingestion_format, message_id, latency_timestamps). - Swap every event["x"] / event.fetch("x") for typed accessors across FailedEventError, IndexingFailuresError, Operation::Factory and Update, DatastoreIndexingRouter, Processor, the JSON ingestion adapter and EnvelopeValidator, WarehouseDumper, TestSupport::Converters, and the affected specs and RBS files. - Event#event_id returns the EventID for a validated event, so callers ask the event for its id rather than building one from its fields. EventID.from_decoded_hash covers the one case with no Event to ask: a payload that failed envelope validation, where any envelope field can be absent. Both keep the "type:id@vversion" format in EventID. - FailedEventError delegates id/op/type/version/record/message_id to its event via Forwardable instead of manual wrapper methods, and includes IndexingFailuresError::_IndexingFailure so Steep enforces the interface it already satisfies. - EnvelopeValidator#events_from now returns Events via Event.from_validated_hash instead of hashes. - TestSupport::Converters keeps upsert_event_hash_for alongside upsert_event_for, since the JSON hash and the Event are now distinct: specs that exercise envelope handling need the hash. - Add a JSONIngestion::Indexer spec proving envelope validation still applies to a type configured to skip record validation, since that guarantee now depends on events_from and validate_event being separate steps rather than an explicit check in the operation factory. Generated with Claude Code

myronmarston added this pull request to stack #1399 September 18, 2026 18:43

This was referenced Sep 18, 2026

Report indexing failures through a _IndexingFailure interface #1395

Merged

Make _RecordPreparer generic #1396

Merged

myronmarston mentioned this pull request Sep 18, 2026

Validate the event envelope before building operations #1397

Open

myronmarston force-pushed the myron/typed-event-object branch from 10c5980 to cdcdefd Compare September 18, 2026 19:32

myronmarston removed this pull request from stack #1399 September 18, 2026 19:44

myronmarston added this pull request to stack #1401 September 18, 2026 19:49

This was referenced Sep 18, 2026

Point test datastore URLs at 127.0.0.1 instead of localhost #1400

Merged

Move record id preparation into the JSON ingestion adapter #1393

Merged

myronmarston force-pushed the myron/typed-event-object branch 2 times, most recently from 98d0305 to 691f157 Compare September 18, 2026 20:41

myronmarston force-pushed the myron/typed-event-object branch 2 times, most recently from 42659c2 to 8e057bf Compare September 18, 2026 23:07

myronmarston force-pushed the myron/typed-event-object branch from 8e057bf to 1c9e973 Compare September 19, 2026 01:33

myronmarston force-pushed the myron/typed-event-object branch from 1c9e973 to 1c1b78b Compare September 20, 2026 01:03

myronmarston marked this pull request as ready for review September 20, 2026 02:18

myronmarston requested review from BrianSigafoos-SQ, bsorbo, ellisandrews-toast, jwils, jwondrusch and marcdaniels-toast as code owners September 20, 2026 02:18

myronmarston requested a review from rossroberts-toast as a code owner September 20, 2026 02:18

myronmarston force-pushed the myron/typed-event-object branch from 1c1b78b to aa3b1f3 Compare September 20, 2026 04:57

jwils approved these changes Sep 21, 2026

View reviewed changes

myronmarston force-pushed the myron/typed-event-object branch from aa3b1f3 to efdd817 Compare September 21, 2026 00:48

jwils mentioned this pull request Sep 21, 2026

Support protobuf ingestion end to end #1388

Draft

Navigation Menu

Route events through a typed Event value - #1398

Route events through a typed Event value#1398
myronmarston wants to merge 1 commit into
myron/validate-envelope-up-frontfrom
myron/typed-event-object

myronmarston commented Sep 18, 2026

Uh oh!

myronmarston commented Sep 18, 2026 •

edited

Loading

Uh oh!

jwils Sep 21, 2026

Uh oh!

jwils Sep 21, 2026

Uh oh!

jwils Sep 21, 2026

Uh oh!

jwils Sep 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

-          "max_widget_cost" => ([w1] + widgets).select { |w| w.record.dig("cost", "currency") == "USD" }.map { |w| w.record.fetch("cost").fetch("amount_cents") }.max
+          "max_widget_cost" => ([w1] + widgets).filter_map do |w|
+            w.record.fetch("cost").fetch("amount_cents") if w.record.dig("cost", "currency") == "USD"
+          end.max

	latency_timestamps = hash["latency_timestamps"] \|\| {} # : ::Hash[::String, ::String]
	hash.fetch("latency_timestamps", {})

		schema_version: hash.fetch(JSON_SCHEMA_VERSION_KEY),
		ingestion_format: hash.fetch(INGESTION_FORMAT_KEY, "json"),

Conversation

myronmarston commented Sep 18, 2026

Why

What

Uh oh!

myronmarston commented Sep 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jwils Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

jwils Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

jwils Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

jwils Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

myronmarston commented Sep 18, 2026 •

edited

Loading