Add psuedo code for the verification loop to explain the algorithm and...

Optionally, the gRPC streaming can be used to fetch records live,

but allowing the streaming replication to resume from the specific cursor.

### **Verification Service Implementation**

### High Level Pseudo Code

```ruby

class ClaimsVerificationService

  CURSOR_RANGE_SIZE = 1000  # Records per TS request

  LOCAL_BATCH_SIZE = 500    # Local records per batch

  RECENT_RECORD_THRESHOLD = 1.hour  # Skip recent records

  def self.verify_all_claim_models(models_with_claims)

    models_with_claims.each do |model_class|

      verify_model_claims(model_class)

    end

  end

# Configuration

BATCH_SIZE = 1000

RECENT_RECORD_THRESHOLD = 1 hour  # Skip recent records to avoid race conditions

  def self.verify_model_claims(model_class)

    topology_service = Gitlab::Cells::TopologyServiceClient.new

# Main verification process - runs for each claimable model/source type

def verify_claims_for_source(source_type)

    cursor = 0

    loop do

      # Fetch claims from TS in cursor-based ranges

      response = topology_service.list_claims(

        ListClaimsRequest.new(

        # Fetch next batch from TS (skips gaps automatically in case where sparse source_ids are present in TS)

        ts_records = TS.ListRecords(

            cell_id: current_cell_id,

          table_name: model_class.cell_cluster_table_name,

          cursor: cursor,

          limit: CURSOR_RANGE_SIZE

        )

            source_type: source_type,

            id_gt: cursor,

            limit: BATCH_SIZE,

        )

      break if response.claims.empty?

      # Process the returned ID range

      process_claims_range(model_class, response.claims, response.start_range, response.end_range, config)

      cursor = response.next_cursor

      break if cursor.nil?

    end

        # If TS has no records in this range, switch to backfill mode

        # This handles both:

        #   - Initial backfill (TS is empty)

        #   - Tail backfill (Rails has more records than TS)

        if ts_records.empty?

            backfill_from_rails(source_type, cursor)

            break

        end

  def self.process_claims_range(model_class, ts_claims, start_range, end_range, config)

    # Create hash map for O(1) claim lookups

    mapped_ts_claims = ts_claims.to_h(&:method(:claim_key))

    recent_threshold = Time.current - RECENT_RECORD_THRESHOLD

    # Query local records matching the exact TS range

    model_class.where(id: start_range...end_range).find_in_batches(batch_size: LOCAL_BATCH_SIZE) do |local_batch|

      missing_ts_claims = []

      different_local_claims = []

      different_ts_claims = []

      local_batch.each do |record|

        # Skip recent records to avoid transient conflicts

        next if record_is_recent?(record, recent_threshold)

        model_class.cell_claim_attributes.each do |claim_attributes|

          # Generate expected claim from local record

          claim_record = record.generate_claim_record(claim_attributes)

          # Look up and remove from hash (tracks processed claims)

          ts_claim_record = mapped_ts_claims.delete(claim_key(claim_record))

          if ! ts_claim_record

            # Missing in TS: local exists, no TS claim

            missing_ts_claims << claim_record

          elsif ! claim_records_match?(claim_record, ts_claim_record)

            # Different: both exist but data differs

            next if ts_claim_is_recent?(ts_claim_record, recent_threshold)

            different_local_claims << claim_record

            different_ts_claims << ts_claim_record

        # Use max ID from TS response to bound Rails query

        # This catches Rails records in any gaps TS skipped over

        max_ts_id = ts_records.max_by(&:source_id).source_id

        # Fetch Rails records in the same ID range

        rails_records = Rails.GetRecordsForSource(source_type, id_gt: cursor, id_lte: max_ts_id)

        # Build lookup maps for O(1) comparison

        ts_map = ts_records.index_by(&:source_id)

        rails_map = rails_records.index_by(&:id)

        # Collect discrepancies by type for batch processing

        missing_in_ts = []      # Exists in Rails, not in TS

        different_records = []  # Exists in both, but data differs

        extra_in_ts = []        # Exists in TS, not in Rails

        # Get union of all IDs from both sources

        all_ids = ts_map.keys | rails_map.keys

        all_ids.each do |id|

            ts_record = ts_map[id]

            rails_record = rails_map[id]

            # Skip recent Rails records to avoid interfering with in-flight transactions

            next if rails_record && rails_record.created_at > (Time.current - RECENT_RECORD_THRESHOLD)

            # Skip recent TS records for the same reason

            next if ts_record && ts_record.created_at > (Time.current - RECENT_RECORD_THRESHOLD)

            if ts_record && rails_record

                unless records_match?(ts_record, rails_record)

                    # Data mismatch - Rails is authoritative

                    different_records << { ts: ts_record, rails: rails_record }

                end

          # If match exactly, no action needed

            elsif rails_record && !ts_record

                # Missing in TS - shouldn't happen during normal operation

                log_exception(source_type, id)

                missing_in_ts << rails_record

            elsif ts_record && !rails_record

                # Orphan in TS - needs cleanup

                extra_in_ts << ts_record

            end

        end

      # Execute corrections in batches

      if missing_ts_claims.present?

        # TODO: This can fail if it's already claimed by the other cell. Need manual intervene.

        begin_update_and_commit(creates: missing_ts_claims)

        # Process discrepancies using BeginUpdate/CommitUpdate pattern

        # Each correction type is processed separately for clarity and error isolation

        if missing_in_ts.any?

            # Create missing records in TS

            # Note: May fail if record was claimed by another cell - requires manual intervention

            TS.begin_update_and_commit(creates: missing_in_ts)

        end

      if different_ts_claims.present?

        # TODO: This can fail if it's already claimed by the other cell. Need manual intervene.

        # Non-atomic update: destroy then create

        begin_update_and_commit(destroys: different_ts_claims)

        begin_update_and_commit(creates: different_local_claims)

        if different_records.any?

            # Update requires destroy + create (no direct update API)

            # Non-atomic: if create fails after destroy, record is lost until next verification

            ts_records_to_destroy = different_records.map { |r| r[:ts] }

            rails_records_to_create = different_records.map { |r| r[:rails] }

            TS.begin_update_and_commit(destroys: ts_records_to_destroy)

            TS.begin_update_and_commit(creates: rails_records_to_create)

        end

        if extra_in_ts.any?

            # Remove orphaned records from TS

            TS.begin_update_and_commit(destroys: extra_in_ts)

        end

    # Remaining hash entries are extra TS claims

    extra_ts_claims = mapped_ts_claims.values.reject { |ts_claim| ts_claim_is_recent?(ts_claim, recent_threshold) }

    if extra_ts_claims.present?

      begin_update_and_commit(destroys: extra_ts_claims)

        # Move to next range

        cursor = max_ts_id

    end

end

  def self.record_is_recent?(record, recent_threshold)

    record.created_at > recent_threshold

  end

  def self.ts_claim_is_recent?(ts_claim, recent_threshold)

    ts_claim.created_at.to_time > recent_threshold

  end

# Backfill function - creates all Rails records beyond start_id in TS

# Used for initial population and catching up when Rails is ahead of TS

def backfill_from_rails(source_type, start_id)

    cursor = start_id

    loop do

        # Fetch Rails records in batches to manage memory

        batch = Rails.GetRecordsForSource(

            source_type,

            id_gt: cursor,

            order_by: :id,

            limit: 1000

        )

        break if batch.empty?

        # Skip recent records during backfill as well

        records_to_create = batch.reject { |r| r.created_at > (Time.current - RECENT_RECORD_THRESHOLD) }

  def self.claim_records_match?(expected_claim, ts_claim)

    # TODO: Compare claim type, value, and owner

        TS.begin_update_and_commit(creates: records_to_create) if records_to_create.any?

        cursor = batch.last.id

    end

end

  def self.claim_key(ts_claim)

    # Unique identifier for hash lookups

    "#{ts_claim.claim_type}/#{ts_claim.claim_value}"

# Helper: Compare TS record with Rails record

# Returns true if all claim-relevant fields match

def records_match?(ts_record, rails_record)

    ts_record.metadata == rails_record.metadata

end

  def self.begin_update_and_commit(creates: [], destroys: [])

    topology_service = Gitlab::Cells::TopologyServiceClient.new

    # Use standard BeginUpdate/CommitUpdate pattern

    response = topology_service.begin_update(

      BeginUpdateRequest.new(

# Helper: BeginUpdate followed by immediate CommitUpdate

# Used by verification since there are no local DB changes to coordinate

def TS.begin_update_and_commit(creates: [], destroys: [])

    response = TS.BeginUpdate(

        cell_id: current_cell_id,

        creates: creates,

        destroys: destroys

    )

    )

    # Immediate commit (no local DB changes for verification)

    topology_service.commit_update(

      CommitUpdateRequest.new(

    TS.CommitUpdate(

        cell_id: current_cell_id,

        lease_uuid: response.lease_payload.lease_uuid

        lease_uuid: response.lease_uuid

    )

    )

    Rails.logger.info "Verified: #{creates.size} creates, #{destroys.size} destroys"

  end

  private

  def self.current_cell_id

    Gitlab::Cells.current_cell_id

  end

end

```