Use Buf to guard protobuf schema compatibility

.github/workflows/ci.yaml

@@ -91,6 +91,13 @@ jobs:
           ruby-version: ${{ matrix.ruby }}
           bundler-cache: true # runs 'bundle install' and caches installed gems automatically
 
+      - name: Setup Buf
+        if: matrix.build_part == 'run_misc_checks'
+        uses: bufbuild/buf-action@8c6a16e16f12ba20b6470afa9c2ba9b5ba8c97c3 # v1.5.0
+        with:
+          version: "1.72.0"
+          setup_only: true
+
       - name: Setup Docker Compose
         uses: KengoTODA/actions-setup-docker-compose@477353946803dd64eaa44008b865b6bfc88cab4e # v1.2.4
         env:

elasticgraph-proto_ingestion/README.md

@@ -77,6 +77,28 @@ end
 After running `bundle exec rake schema_artifacts:dump`, ElasticGraph will generate a `schema.proto`
 schema artifact, and will maintain a `proto_field_numbers.yaml` file alongside your schema definition.
 
+### Protecting Protobuf Compatibility With Buf
+
+Install the [Buf CLI](https://buf.build/docs/installation/) anywhere that dumps schema artifacts.
+Once `schema.proto` exists, ElasticGraph compiles the existing and proposed schemas with Buf whenever
+the artifact changes. It then applies Buf's strict [`FILE` breaking-change
+rules](https://buf.build/docs/breaking/rules/) so changes that break generated code, JSON, or the binary
+wire format cannot be dumped accidentally. Imported schemas are compiled using the project's normal
+Buf configuration and are excluded from the comparison itself, keeping this check scoped to the
+generated `schema.proto`.
+
+A breaking change always fails the dump. Protobuf has no schema version to bump: consumers that
+already deserialize these messages would misread them, and you cannot update every consumer at the
+same moment. So there is no flag to accept a breaking change.
+
+To get past the error, make the change compatible:
+
+- Add a new field instead of a change to the type or the name of an existing field.
+- Reserve the number of every field you remove.
+- Keep enum value numbers stable, and reserve the numbers of values you remove.
+
+Compatible additions dump normally and need no extra step.
+
 ## Schema Definition API
 
 ### Protobuf Syntax (`proto2` / `proto3`)

elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/buf_breaking_change_detector.rb

@@ -0,0 +1,99 @@
+# Copyright 2024 - 2026 Block, Inc.
+#
+# Use of this source code is governed by an MIT-style
+# license that can be found in the LICENSE file or at
+# https://opensource.org/licenses/MIT.
+#
+# frozen_string_literal: true
+
+require "elastic_graph/errors"
+require "elastic_graph/proto_ingestion"
+require "open3"
+require "tempfile"
+
+module ElasticGraph
+  module ProtoIngestion
+    module SchemaDefinition
+      # Uses the Buf CLI to detect breaking changes between two protobuf schemas.
+      #
+      # @private
+      class BufBreakingChangeDetector
+        BREAKING_CONFIG_JSON = %({"version":"v2","breaking":{"use":["FILE"]}})
+
+        def initialize(buf_command: "buf", temporary_directory: ::Dir.pwd)
+          @buf_command = buf_command
+          @temporary_directory = temporary_directory
+        end
+
+        # Returns Buf's diagnostics if the current schema breaks the prior schema, or `nil` if it
+        # is compatible.
+        def breaking_changes(current_schema:, against_schema:)
+          with_built_images(current_schema, against_schema) do |current_image_path, against_image_path, proto_path|
+            output, status = run_buf(
+              "breaking",
+              current_image_path,
+              "--against",
+              against_image_path,
+              "--exclude-imports",
+              "--config",
+              BREAKING_CONFIG_JSON
+            )
+
+            return nil if status.success?
+            return normalized_output(output, proto_path) if status.exitstatus == 100
+
+            raise Errors::SchemaError, buf_failure_message("compare the protobuf schemas", output)
+          end
+        rescue ::Errno::ENOENT
+          raise Errors::SchemaError, <<~EOS.strip
+            The Buf CLI is required to check `schema.proto` for breaking changes, but the `#{@buf_command}` command could not be found.
+            Install Buf from https://buf.build/docs/installation and run `bundle exec rake schema_artifacts:dump` again.
+          EOS
+        end
+
+        private
+
+        def with_built_images(current_schema, against_schema)
+          ::Tempfile.create(["elasticgraph-schema", ".proto"], @temporary_directory) do |proto_file|
+            ::Tempfile.create(["elasticgraph-current-schema", ".binpb"], @temporary_directory) do |current_image_file|
+              build_image(current_schema, "current", proto_file, current_image_file)
+
+              ::Tempfile.create(["elasticgraph-against-schema", ".binpb"], @temporary_directory) do |against_image_file|
+                build_image(against_schema, "against", proto_file, against_image_file)
+                yield current_image_file.path, against_image_file.path, proto_file.path
+              end
+            end
+          end
+        end
+
+        def build_image(schema, label, proto_file, image_file)
+          proto_file.rewind
+          proto_file.truncate(0)
+          proto_file.write(schema)
+          proto_file.flush
+          image_file.close
+
+          output, status = run_buf("build", proto_file.path, "--output", image_file.path)
+          return if status.success?
+
+          raise Errors::SchemaError, buf_failure_message("compile the #{label} protobuf schema", output)
+        end
+
+        def run_buf(*arguments)
+          stdout, stderr, status = ::Open3.capture3(@buf_command, *arguments)
+          output = [stdout, stderr].reject(&:empty?).join("\n").strip
+          [output, status]
+        end
+
+        def buf_failure_message(action, output)
+          details = output.empty? ? "Buf did not provide any diagnostics." : output
+          "Buf was unable to #{action}:\n\n#{details}"
+        end
+
+        def normalized_output(output, proto_path)
+          output.gsub(proto_path, PROTO_SCHEMA_FILE).gsub(::File.basename(proto_path), PROTO_SCHEMA_FILE)
+        end
+      end
+    end
+  end
+end

elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema_artifact_manager_extension.rb

@@ -8,6 +8,7 @@
 
 require "elastic_graph/errors"
 require "elastic_graph/proto_ingestion"
+require "elastic_graph/proto_ingestion/schema_definition/buf_breaking_change_detector"
 
 module ElasticGraph
   module ProtoIngestion
@@ -31,6 +32,24 @@ module SchemaArtifactManagerExtension
           "lost, and previously serialized protobuf messages would be misread."
         ].freeze
 
+        # Overrides `dump_artifacts` to reject any change that breaks protobuf wire compatibility.
+        #
+        # A protobuf schema has no version to bump. Every consumer that already deserializes these
+        # messages must keep reading them, and those consumers cannot be updated in lockstep. So a
+        # breaking change is always an error, and the only fix is to make the change compatible.
+        def dump_artifacts
+          proto_schema = protobuf_schema_definition_results.proto_schema
+          return super if proto_schema.empty?
+
+          artifact = proto_schema_artifact
+          existing_schema = artifact.existing_dumped_contents
+          return super unless artifact.out_of_date? && existing_schema
+
+          check_for_breaking_proto_changes(artifact.desired_contents, existing_schema)
+
+          super
+        end
+
         private
 
         # Overrides the base `artifacts_from_schema_def` method to add proto artifacts.
@@ -41,10 +60,41 @@ def artifacts_from_schema_def
 
           base_artifacts + [
             proto_field_numbers_artifact,
-            new_raw_artifact(PROTO_SCHEMA_FILE, proto_schema.chomp, comment_prefix: "//")
+            proto_schema_artifact
           ]
         end
 
+        def proto_schema_artifact
+          @proto_schema_artifact ||= new_raw_artifact(
+            PROTO_SCHEMA_FILE,
+            protobuf_schema_definition_results.proto_schema.chomp,
+            comment_prefix: "//"
+          )
+        end
+
+        def check_for_breaking_proto_changes(current_schema, against_schema)
+          changes = BufBreakingChangeDetector.new(
+            temporary_directory: ::File.dirname(proto_schema_artifact.file_name)
+          ).breaking_changes(
+            current_schema: current_schema,
+            against_schema: against_schema
+          )
+          return unless changes
+
+          abort <<~EOS.strip
+            Buf detected a breaking change to `schema.proto`:
+
+            #{changes}
+
+            Protobuf offers no way to version your way out of this. Consumers that already read
+            these messages would misread them after this change. Make the change compatible
+            instead: add a new field rather than retype or rename an existing one, and reserve the
+            number of every field you remove.
+          EOS
+        rescue Errors::SchemaError => e
+          abort e.message
+        end
+
         # Builds the `proto_field_numbers.yaml` artifact. The file is part of the schema definition
         # rather than a proper schema artifact--it's an input to `schema.proto` generation--so it
         # lives alongside `path_to_schema` instead of in the schema artifacts directory.

elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/buf_breaking_change_detector.rbs

@@ -0,0 +1,23 @@
+module ElasticGraph
+  module ProtoIngestion
+    module SchemaDefinition
+      class BufBreakingChangeDetector
+        BREAKING_CONFIG_JSON: ::String
+
+        @buf_command: ::String
+        @temporary_directory: ::String
+
+        def initialize: (?buf_command: ::String, ?temporary_directory: ::String) -> void
+        def breaking_changes: (current_schema: ::String, against_schema: ::String) -> (::String | nil)
+
+        private
+
+        def with_built_images: (::String current_schema, ::String against_schema) { (::String, ::String, ::String) -> untyped } -> untyped
+        def build_image: (::String schema, ::String label, ::File proto_file, ::File image_file) -> void
+        def run_buf: (*::String arguments) -> [::String, ::Process::Status]
+        def buf_failure_message: (::String action, ::String output) -> ::String
+        def normalized_output: (::String, ::String) -> ::String
+      end
+    end
+  end
+end

elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema_artifact_manager_extension.rbs

@@ -4,9 +4,15 @@ module ElasticGraph
       module SchemaArtifactManagerExtension : ::ElasticGraph::SchemaDefinition::SchemaArtifactManager
         PROTO_FIELD_NUMBERS_COMMENT_PREAMBLE_LINES: ::Array[::String]
 
+        @proto_schema_artifact: (::ElasticGraph::SchemaDefinition::SchemaArtifact[::String] | nil)
+
+        def dump_artifacts: () -> void
+
         private
 
         def artifacts_from_schema_def: () -> ::Array[::ElasticGraph::SchemaDefinition::SchemaArtifact[untyped]]
+        def proto_schema_artifact: () -> ::ElasticGraph::SchemaDefinition::SchemaArtifact[::String]
+        def check_for_breaking_proto_changes: (::String, ::String) -> void
         def proto_field_numbers_artifact: () -> ::ElasticGraph::SchemaDefinition::SchemaArtifact[::Hash[::String, untyped]]
         def proto_field_numbers_path: () -> ::String
         def protobuf_schema_definition_results: () -> (::ElasticGraph::SchemaDefinition::Results & ResultsExtension)

elasticgraph-proto_ingestion/spec/integration/elastic_graph/proto_ingestion/schema_definition/rake_tasks_spec.rb

@@ -38,6 +38,25 @@ module SchemaDefinition
               )
           end
 
+          it "does not invoke Buf or dump a proto artifact when no indexed types are defined" do
+            expect(BufBreakingChangeDetector).not_to receive(:new)
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Point" do |t|
+                t.field "x", "Float"
+              end
+
+              s.on_root_query_type do |t|
+                t.field "point", "Point" do |f|
+                  f.resolve_with :object_without_lookahead
+                end
+              end
+            EOS
+
+            run_rake_with_proto("schema_artifacts:dump")
+
+            expect(read_artifact(PROTO_SCHEMA_FILE)).to be_nil
+          end
+
           it "idempotently dumps proto artifacts" do
             write_proto_schema(table_defs: <<~EOS)
               s.object_type "Product" do |t|
@@ -55,6 +74,8 @@ module SchemaDefinition
           end
 
           it "persists proto field-number mappings and reuses them on the next dump" do
+            stub_buf_breaking_changes(nil)
+
             write_proto_schema(table_defs: <<~EOS)
               s.object_type "Product" do |t|
                 t.field "id", "ID"
@@ -96,6 +117,88 @@ module SchemaDefinition
 
             expect(read_artifact(PROTO_SCHEMA_FILE)).to include("string id = 1;", "string name = 2;")
           end
+
+          it "dumps a compatible change that Buf reports as safe" do
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "ID"
+                t.index "products"
+              end
+            EOS
+            run_rake_with_proto("schema_artifacts:dump")
+
+            detector = stub_buf_breaking_changes(nil)
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "ID"
+                t.field "name", "String"
+                t.index "products"
+              end
+            EOS
+
+            expect {
+              run_rake_with_proto("schema_artifacts:dump")
+            }.to change { read_artifact(PROTO_SCHEMA_FILE) }
+
+            expect(detector).to have_received(:breaking_changes).with(
+              current_schema: a_string_including("string name = 2;"),
+              against_schema: a_string_excluding("string name = 2;")
+            )
+          end
+
+          it "refuses to dump a breaking change reported by Buf" do
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "ID"
+                t.index "products"
+              end
+            EOS
+            run_rake_with_proto("schema_artifacts:dump")
+
+            original_proto = read_artifact(PROTO_SCHEMA_FILE)
+            stub_buf_breaking_changes("schema.proto: Field changed type from string to int32.")
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "Int"
+                t.index "products"
+              end
+            EOS
+
+            expect {
+              run_rake_with_proto("schema_artifacts:dump")
+            }.to abort_with a_string_including(
+              "Buf detected a breaking change",
+              "Field changed type from string to int32",
+              "Protobuf offers no way to version your way out of this",
+              "reserve the"
+            )
+            expect(read_artifact(PROTO_SCHEMA_FILE)).to eq(original_proto)
+          end
+
+          it "surfaces Buf failures without a stack trace" do
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "ID"
+                t.index "products"
+              end
+            EOS
+            run_rake_with_proto("schema_artifacts:dump")
+
+            detector = instance_double(BufBreakingChangeDetector)
+            allow(detector).to receive(:breaking_changes).and_raise(Errors::SchemaError, "Buf could not compile an import.")
+            allow(BufBreakingChangeDetector).to receive(:new).and_return(detector)
+            write_proto_schema(table_defs: <<~EOS)
+              s.object_type "Product" do |t|
+                t.field "id", "ID"
+                t.field "name", "String"
+                t.index "products"
+              end
+            EOS
+
+            expect {
+              run_rake_with_proto("schema_artifacts:dump")
+            }.to abort_with("Buf could not compile an import.")
+          end
         end
 
         describe "schema_artifacts:check" do
@@ -129,6 +232,12 @@ def write_proto_schema(table_defs:)
           EOS
         end
 
+        def stub_buf_breaking_changes(changes)
+          detector = instance_double(BufBreakingChangeDetector, breaking_changes: changes)
+          allow(BufBreakingChangeDetector).to receive(:new).and_return(detector)
+          detector
+        end
+
         def run_rake_with_proto(*args)
           run_rake(*args) do |output|
             ElasticGraph::SchemaDefinition::RakeTasks.new(