Add federation ingestion limits and tests (#364)

web/lib/potato_mesh/application/federation.rb

@@ -445,12 +445,34 @@ module PotatoMesh
       # @param db [SQLite3::Database] open database connection used for writes.
       # @param domain [String] remote domain to crawl for federation records.
       # @param visited [Set<String>] domains processed during this crawl.
+      # @param per_response_limit [Integer, nil] maximum entries processed per response.
+      # @param overall_limit [Integer, nil] maximum unique domains visited.
       # @return [Set<String>] updated set of visited domains.
-      def ingest_known_instances_from!(db, domain, visited: nil)
+      def ingest_known_instances_from!(
+        db,
+        domain,
+        visited: nil,
+        per_response_limit: nil,
+        overall_limit: nil
+      )
         sanitized = sanitize_instance_domain(domain)
         return visited || Set.new unless sanitized
 
         visited ||= Set.new
+
+        overall_limit ||= PotatoMesh::Config.federation_max_domains_per_crawl
+        per_response_limit ||= PotatoMesh::Config.federation_max_instances_per_response
+
+        if overall_limit && overall_limit.positive? && visited.size >= overall_limit
+          debug_log(
+            "Skipped remote instance crawl due to crawl limit",
+            context: "federation.instances",
+            domain: sanitized,
+            limit: overall_limit,
+          )
+          return visited
+        end
+
         return visited if visited.include?(sanitized)
 
         visited << sanitized
@@ -466,7 +488,29 @@ module PotatoMesh
           return visited
         end
 
+        processed_entries = 0
         payload.each do |entry|
+          if per_response_limit && per_response_limit.positive? && processed_entries >= per_response_limit
+            debug_log(
+              "Skipped remote instance entry due to response limit",
+              context: "federation.instances",
+              domain: sanitized,
+              limit: per_response_limit,
+            )
+            break
+          end
+
+          if overall_limit && overall_limit.positive? && visited.size >= overall_limit
+            debug_log(
+              "Skipped remote instance entry due to crawl limit",
+              context: "federation.instances",
+              domain: sanitized,
+              limit: overall_limit,
+            )
+            break
+          end
+
+          processed_entries += 1
           attributes, signature, reason = remote_instance_attributes_from_payload(entry)
           unless attributes && signature
             warn_log(
@@ -523,7 +567,13 @@ module PotatoMesh
 
           begin
             upsert_instance_record(db, attributes, signature)
-            ingest_known_instances_from!(db, attributes[:domain], visited: visited)
+            ingest_known_instances_from!(
+              db,
+              attributes[:domain],
+              visited: visited,
+              per_response_limit: per_response_limit,
+              overall_limit: overall_limit,
+            )
           rescue ArgumentError => e
             warn_log(
               "Failed to persist remote instance",

web/lib/potato_mesh/application/routes/ingest.rb

@@ -225,7 +225,12 @@ module PotatoMesh
 
             db = open_database
             upsert_instance_record(db, attributes, signature)
-            ingest_known_instances_from!(db, attributes[:domain])
+            ingest_known_instances_from!(
+              db,
+              attributes[:domain],
+              per_response_limit: PotatoMesh::Config.federation_max_instances_per_response,
+              overall_limit: PotatoMesh::Config.federation_max_domains_per_crawl,
+            )
             debug_log(
               "Registered remote instance",
               context: "ingest.register",

web/lib/potato_mesh/config.rb

@@ -34,6 +34,8 @@ module PotatoMesh
     DEFAULT_MAX_DISTANCE_KM = 42.0
     DEFAULT_REMOTE_INSTANCE_CONNECT_TIMEOUT = 5
     DEFAULT_REMOTE_INSTANCE_READ_TIMEOUT = 12
+    DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE = 64
+    DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL = 256
 
     # Resolve the absolute path to the web application root directory.
     #
@@ -285,6 +287,26 @@ module PotatoMesh
       DEFAULT_REMOTE_INSTANCE_READ_TIMEOUT
     end
 
+    # Limit the number of remote instances processed from a single response.
+    #
+    # @return [Integer] maximum entries processed per /api/instances payload.
+    def federation_max_instances_per_response
+      fetch_positive_integer(
+        "FEDERATION_MAX_INSTANCES_PER_RESPONSE",
+        DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE,
+      )
+    end
+
+    # Limit the total number of distinct domains crawled during one ingestion.
+    #
+    # @return [Integer] maximum unique domains visited per crawl.
+    def federation_max_domains_per_crawl
+      fetch_positive_integer(
+        "FEDERATION_MAX_DOMAINS_PER_CRAWL",
+        DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL,
+      )
+    end
+
     # Maximum acceptable age for remote node data.
     #
     # @return [Integer] seconds before remote nodes are considered stale.
@@ -424,6 +446,27 @@ module PotatoMesh
       trimmed.empty? ? default : trimmed
     end
 
+    # Fetch and validate integer based configuration flags.
+    #
+    # @param key [String] environment variable to read.
+    # @param default [Integer] fallback value when unset or invalid.
+    # @return [Integer] positive integer sourced from configuration.
+    def fetch_positive_integer(key, default)
+      value = ENV[key]
+      return default if value.nil?
+
+      trimmed = value.strip
+      return default if trimmed.empty?
+
+      begin
+        parsed = Integer(trimmed, 10)
+      rescue ArgumentError
+        return default
+      end
+
+      parsed.positive? ? parsed : default
+    end
+
     # Resolve the effective XDG directory honoring environment overrides.
     #
     # @param env_key [String] name of the environment variable to inspect.

web/spec/config_spec.rb

@@ -169,6 +169,54 @@ RSpec.describe PotatoMesh::Config do
     end
   end
 
+  describe ".federation_max_instances_per_response" do
+    it "returns the baked-in response limit when unset" do
+      within_env("FEDERATION_MAX_INSTANCES_PER_RESPONSE" => nil) do
+        expect(described_class.federation_max_instances_per_response).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_MAX_INSTANCES_PER_RESPONSE" => "7") do
+        expect(described_class.federation_max_instances_per_response).to eq(7)
+      end
+    end
+
+    it "rejects non-positive overrides" do
+      within_env("FEDERATION_MAX_INSTANCES_PER_RESPONSE" => "0") do
+        expect(described_class.federation_max_instances_per_response).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_MAX_INSTANCES_PER_RESPONSE,
+        )
+      end
+    end
+  end
+
+  describe ".federation_max_domains_per_crawl" do
+    it "returns the baked-in crawl limit when unset" do
+      within_env("FEDERATION_MAX_DOMAINS_PER_CRAWL" => nil) do
+        expect(described_class.federation_max_domains_per_crawl).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL,
+        )
+      end
+    end
+
+    it "accepts positive overrides" do
+      within_env("FEDERATION_MAX_DOMAINS_PER_CRAWL" => "11") do
+        expect(described_class.federation_max_domains_per_crawl).to eq(11)
+      end
+    end
+
+    it "rejects invalid overrides" do
+      within_env("FEDERATION_MAX_DOMAINS_PER_CRAWL" => "-5") do
+        expect(described_class.federation_max_domains_per_crawl).to eq(
+          PotatoMesh::Config::DEFAULT_FEDERATION_MAX_DOMAINS_PER_CRAWL,
+        )
+      end
+    end
+  end
+
   describe ".db_path" do
     it "returns the default path inside the data directory" do
       expect(described_class.db_path).to eq(described_class.default_db_path)

web/spec/federation_spec.rb

@@ -15,6 +15,7 @@
 require "spec_helper"
 require "net/http"
 require "openssl"
+require "set"
 require "uri"
 
 RSpec.describe PotatoMesh::App::Federation do
@@ -177,6 +178,97 @@ RSpec.describe PotatoMesh::App::Federation do
     end
   end
 
+  describe ".ingest_known_instances_from!" do
+    let(:db) { double(:db) }
+    let(:seed_domain) { "seed.mesh" }
+    let(:payload_entries) do
+      Array.new(3) do |index|
+        {
+          "id" => "remote-#{index}",
+          "domain" => "ally-#{index}.mesh",
+          "pubkey" => "ignored-pubkey-#{index}",
+          "signature" => "ignored-signature-#{index}",
+        }
+      end
+    end
+    let(:attributes_list) do
+      payload_entries.map do |entry|
+        {
+          id: entry["id"],
+          domain: entry["domain"],
+          pubkey: entry["pubkey"],
+          name: nil,
+          version: nil,
+          channel: nil,
+          frequency: nil,
+          latitude: nil,
+          longitude: nil,
+          last_update_time: nil,
+          is_private: false,
+        }
+      end
+    end
+    let(:node_payload) do
+      Array.new(PotatoMesh::Config.remote_instance_min_node_count) do |index|
+        { "node_id" => "node-#{index}", "last_heard" => Time.now.to_i - index }
+      end
+    end
+    let(:response_map) do
+      mapping = { [seed_domain, "/api/instances"] => [payload_entries, :instances] }
+      attributes_list.each do |attributes|
+        mapping[[attributes[:domain], "/api/nodes"]] = [node_payload, :nodes]
+        mapping[[attributes[:domain], "/api/instances"]] = [[], :instances]
+      end
+      mapping
+    end
+
+    before do
+      allow(federation_helpers).to receive(:fetch_instance_json) do |host, path|
+        response_map.fetch([host, path]) { [nil, []] }
+      end
+      allow(federation_helpers).to receive(:verify_instance_signature).and_return(true)
+      allow(federation_helpers).to receive(:validate_remote_nodes).and_return([true, nil])
+      payload_entries.each_with_index do |entry, index|
+        allow(federation_helpers).to receive(:remote_instance_attributes_from_payload).with(entry).and_return([attributes_list[index], "signature-#{index}", nil])
+      end
+    end
+
+    it "stops processing once the per-response limit is exceeded" do
+      processed_domains = []
+      allow(federation_helpers).to receive(:upsert_instance_record) do |_db, attrs, _signature|
+        processed_domains << attrs[:domain]
+      end
+      allow(PotatoMesh::Config).to receive(:federation_max_instances_per_response).and_return(2)
+      allow(PotatoMesh::Config).to receive(:federation_max_domains_per_crawl).and_return(10)
+
+      visited = federation_helpers.ingest_known_instances_from!(db, seed_domain)
+
+      expect(processed_domains).to eq([
+        attributes_list[0][:domain],
+        attributes_list[1][:domain],
+      ])
+      expect(visited).to include(seed_domain, attributes_list[0][:domain], attributes_list[1][:domain])
+      expect(visited).not_to include(attributes_list[2][:domain])
+      expect(federation_helpers.debug_messages).to include(a_string_including("response limit"))
+    end
+
+    it "halts recursion once the crawl limit would be exceeded" do
+      processed_domains = []
+      allow(federation_helpers).to receive(:upsert_instance_record) do |_db, attrs, _signature|
+        processed_domains << attrs[:domain]
+      end
+      allow(PotatoMesh::Config).to receive(:federation_max_instances_per_response).and_return(5)
+      allow(PotatoMesh::Config).to receive(:federation_max_domains_per_crawl).and_return(2)
+
+      visited = federation_helpers.ingest_known_instances_from!(db, seed_domain)
+
+      expect(processed_domains).to eq([attributes_list.first[:domain]])
+      expect(visited).to include(seed_domain, attributes_list.first[:domain])
+      expect(visited).not_to include(attributes_list[1][:domain], attributes_list[2][:domain])
+      expect(federation_helpers.debug_messages).to include(a_string_including("crawl limit"))
+    end
+  end
+
   describe ".perform_instance_http_request" do
     let(:uri) { URI.parse("https://remote.example.com/api") }
     let(:http_client) { instance_double(Net::HTTP) }