Sync performance fixes (#1921)

We discovered that the perf improvements for block level sync tend to slow down FG syncing. We reverted the improvements (#1920) to unblock deploys while a fix was figured out.

This combines the reverted perf improvements (#1898 and #1897) and fixes them to be FG compatible.

Gemfile

@@ -43,6 +43,7 @@ gem "kaminari"
 gem "lodash-rails"
 gem "lograge"
 gem "ougai"
+gem "oj"
 gem "parallel", require: false
 gem "passenger"
 gem "pg", ">= 0.18", "< 2.0"

Gemfile.lock

@@ -608,6 +608,7 @@ DEPENDENCIES
   memery
   memory_profiler
   mock_redis
+  oj
   ougai
   parallel
   parallel_tests

app/controllers/api/v3/blood_sugars_controller.rb

@@ -12,7 +12,7 @@ def sync_to_user
 
   private
 
-  def region_records
+  def model_sync_scope
     super.for_v3
   end
 

app/controllers/api/v3/facilities_controller.rb

@@ -16,7 +16,7 @@ def current_facility_records
   end
 
   def other_facility_records
-    Statsd.instance.time("other_facility_records.Facility") do
+    time(__method__) do
       Facility
         .with_discarded
         .updated_on_server_since(other_facilities_processed_since, limit)
@@ -45,7 +45,7 @@ def force_resync?
   end
 
   def records_to_sync
-    Statsd.instance.time("records_to_sync.Facility") do
+    time(__method__) do
       other_facility_records
         .with_block_region_id
         .includes(:facility_group)

app/controllers/api/v3/patients_controller.rb

@@ -13,25 +13,25 @@ def request_metadata
     {request_user_id: current_user.id, request_facility_id: current_facility.id}
   end
 
-  def region_records
-    super.includes(:address, :phone_numbers, :business_identifiers)
-  end
-
   def current_facility_records
-    Statsd.instance.time("current_facility_records.Patient") do
-      region_records
-        .where(registration_facility: current_facility)
-        .updated_on_server_since(current_facility_processed_since, limit)
+    time(__method__) do
+      @current_facility_records ||=
+        current_facility
+          .prioritized_patients
+          .for_sync
+          .updated_on_server_since(current_facility_processed_since, limit)
     end
   end
 
   def other_facility_records
-    Statsd.instance.time("other_facility_records.Patient") do
-      other_facilities_limit = limit - current_facility_records.count
-
-      region_records
-        .where.not(registration_facility: current_facility)
-        .updated_on_server_since(other_facilities_processed_since, other_facilities_limit)
+    time(__method__) do
+      other_facilities_limit = limit - current_facility_records.size
+      @other_facility_records ||=
+        current_sync_region
+          .syncable_patients
+          .where.not(registration_facility: current_facility)
+          .for_sync
+          .updated_on_server_since(other_facilities_processed_since, other_facilities_limit)
     end
   end
 

app/controllers/api/v3/protocols_controller.rb

@@ -16,7 +16,7 @@ def current_facility_records
   end
 
   def other_facility_records
-    Statsd.instance.time("other_facility_records.Protocol") do
+    time(__method__) do
       Protocol
         .with_discarded
         .updated_on_server_since(other_facilities_processed_since, limit)

app/controllers/api/v3/sync_controller.rb

@@ -15,12 +15,11 @@ def __sync_to_user__(response_key)
 
     AuditLog.create_logs_async(current_user, records, "fetch", Time.current) unless disable_audit_logs?
     log_block_level_sync_metrics(response_key)
-
     render(
-      json: {
+      json: Oj.dump({
         response_key => records.map { |record| transform_to_response(record) },
         "process_token" => encode_process_token(response_process_token)
-      },
+      }, mode: :compat),
       status: :ok
     )
   end

app/controllers/api_controller.rb

@@ -68,16 +68,16 @@ def requested_sync_region_id
   end
 
   def validate_facility
-    return head :bad_request unless current_facility.present?
+    head :bad_request unless current_facility.present?
   end
 
   def validate_current_facility_belongs_to_users_facility_group
-    return head :unauthorized unless current_user.present? &&
+    head :unauthorized unless current_user.present? &&
       current_facility_group.facilities.where(id: current_facility.id).present?
   end
 
   def current_user_present?
-    return head :unauthorized unless current_user.present?
+    head :unauthorized unless current_user.present?
   end
 
   def validate_sync_approval_status_allowed

app/controllers/concerns/api/v3/sync_to_user.rb

@@ -2,46 +2,48 @@ module Api::V3::SyncToUser
   extend ActiveSupport::Concern
 
   included do
-    def region_records
-      model.syncable_to_region(current_sync_region)
-    end
-
     def current_facility_records
-      Statsd.instance.time("current_facility_records.#{model.name}") do
-        region_records
-          .where(patient: prioritized_patients)
-          .updated_on_server_since(current_facility_processed_since, limit)
+      time(__method__) do
+        @current_facility_records ||=
+          model_sync_scope
+            .where(patient: current_facility.prioritized_patients.select(:id))
+            .updated_on_server_since(current_facility_processed_since, limit)
       end
     end
 
+    # this is performance-critical code, be careful while refactoring it
     def other_facility_records
-      Statsd.instance.time("other_facility_records.#{model.name}") do
-        other_facilities_limit = limit - current_facility_records.count
-
-        region_records
-          .where.not(patient: prioritized_patients)
-          .updated_on_server_since(other_facilities_processed_since, other_facilities_limit)
+      time(__method__) do
+        other_facilities_limit = limit - current_facility_records.size
+        @other_facility_records ||=
+          model_sync_scope
+            .where("patient_id = ANY (array(?))",
+              current_sync_region
+                .syncable_patients
+                .where.not(registration_facility: current_facility)
+                .select(:id))
+            .updated_on_server_since(other_facilities_processed_since, other_facilities_limit)
       end
     end
 
     private
 
+    def model_sync_scope
+      model.for_sync
+    end
+
     def model
       controller_name.classify.constantize
     end
 
     def records_to_sync
-      Statsd.instance.time("records_to_sync.#{model.name}") do
+      time(__method__) do
         current_facility_records + other_facility_records
       end
     end
 
-    def prioritized_patients
-      current_facility.registered_patients.with_discarded
-    end
-
     def processed_until(records)
-      records.last.updated_at.strftime(APIController::TIME_WITHOUT_TIMEZONE_FORMAT) if records.present?
+      records.last.updated_at.strftime(APIController::TIME_WITHOUT_TIMEZONE_FORMAT) if records.any?
     end
 
     def response_process_token
@@ -90,5 +92,13 @@ def sync_region_modified?
       return if process_token[:sync_region_id].blank?
       process_token[:sync_region_id] != requested_sync_region_id
     end
+
+    def time(method_name, &block)
+      raise ArgumentError, "You must supply a block" unless block
+
+      Statsd.instance.time("#{method_name}.#{model.name}") do
+        yield(block)
+      end
+    end
   end
 end

app/models/appointment.rb

@@ -40,9 +40,7 @@ class Appointment < ApplicationRecord
   validates :device_created_at, presence: true
   validates :device_updated_at, presence: true
 
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   def self.all_overdue
     where(status: "scheduled")

app/models/blood_pressure.rb

@@ -32,9 +32,7 @@ class BloodPressure < ApplicationRecord
       .where(arel_table[:diastolic].lt(THRESHOLDS[:hypertensive][:diastolic]))
   }
 
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   def critical?
     systolic >= THRESHOLDS[:critical][:systolic] || diastolic >= THRESHOLDS[:critical][:diastolic]

app/models/blood_sugar.rb

@@ -30,10 +30,7 @@ class BloodSugar < ApplicationRecord
   V3_TYPES = %i[random post_prandial fasting].freeze
 
   scope :for_v3, -> { where(blood_sugar_type: V3_TYPES) }
-
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   THRESHOLDS = {
     high: {random: 300,

app/models/encounter.rb

@@ -9,9 +9,7 @@ class Encounter < ApplicationRecord
   has_many :blood_pressures, through: :observations, source: :observable, source_type: "BloodPressure"
   has_many :blood_sugars, through: :observations, source: :observable, source_type: "BloodSugar"
 
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   def self.generate_id(facility_id, patient_id, encountered_on)
     UUIDTools::UUID

app/models/facility.rb

@@ -215,6 +215,10 @@ def valid_block
     end
   end
 
+  def prioritized_patients
+    registered_patients.with_discarded
+  end
+
   def self.localized_facility_size(facility_size)
     return unless facility_size
     I18n.t("activerecord.facility.facility_size.#{facility_size}", default: facility_size.capitalize)

app/models/medical_history.rb

@@ -29,9 +29,7 @@ class MedicalHistory < ApplicationRecord
   enum hypertension: MEDICAL_HISTORY_ANSWERS, _prefix: true
   enum diagnosed_with_hypertension: MEDICAL_HISTORY_ANSWERS, _prefix: true
 
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   def indicates_hypertension_risk?
     prior_heart_attack_boolean || prior_stroke_boolean

app/models/patient.rb

@@ -56,7 +56,8 @@ class Patient < ApplicationRecord
 
   attribute :call_result, :string
 
-  scope :syncable_to_region, ->(region) { region.syncable_patients }
+  scope :with_nested_sync_resources, -> { includes(:address, :phone_numbers, :business_identifiers) }
+  scope :for_sync, -> { with_discarded.with_nested_sync_resources }
   scope :search_by_address, ->(term) { joins(:address).merge(Address.search_by_street_or_village(term)) }
   scope :with_diabetes, -> { joins(:medical_history).merge(MedicalHistory.diabetes_yes).distinct }
   scope :with_hypertension, -> { joins(:medical_history).merge(MedicalHistory.hypertension_yes).distinct }

app/models/prescription_drug.rb

@@ -21,9 +21,7 @@ class PrescriptionDrug < ApplicationRecord
   validates :is_protocol_drug, inclusion: {in: [true, false]}
   validates :is_deleted, inclusion: {in: [true, false]}
 
-  scope :syncable_to_region, ->(region) {
-    with_discarded.where(patient: Patient.syncable_to_region(region))
-  }
+  scope :for_sync, -> { with_discarded }
 
   def self.prescribed_as_of(date)
     where("device_created_at <= ?", date.end_of_day)

app/models/region.rb

@@ -100,10 +100,10 @@ def syncable_patients
     case region_type
       when "block"
         registered_patients.with_discarded
-          .union(assigned_patients.with_discarded)
+          .or(assigned_patients.with_discarded)
           .union(appointed_patients.with_discarded)
       else
-        registered_patients
+        registered_patients.with_discarded
     end
   end
 

app/transformers/api/v3/facility_transformer.rb

@@ -7,7 +7,9 @@ def to_response(facility)
           "enable_teleconsultation",
           "teleconsultation_phone_number",
           "teleconsultation_isd_code",
-          "teleconsultation_phone_numbers")
+          "teleconsultation_phone_numbers",
+          "organization_name",
+          "facility_group_name")
         .merge(config: {enable_diabetes_management: facility.enable_diabetes_management,
                         enable_teleconsultation: facility.enable_teleconsultation},
                protocol_id: facility.protocol.try(:id))

app/transformers/api/v3/patient_transformer.rb

@@ -59,10 +59,10 @@ def from_nested_request(payload_attributes)
 
     def to_nested_response(patient)
       Api::V3::Transformer.to_response(patient)
-        .except("address_id")
-        .except("registration_user_id")
-        .except("test_data")
-        .except("deleted_by_user_id")
+        .except("address_id",
+          "registration_user_id",
+          "test_data",
+          "deleted_by_user_id")
         .merge(
           "address" => Api::V3::Transformer.to_response(patient.address),
           "phone_numbers" => patient.phone_numbers.map do |phone_number|

app/transformers/api/v3/transformer.rb

@@ -1,31 +1,33 @@
 class Api::V3::Transformer
   class << self
-    def from_request(attributes_of_payload)
-      rename_attributes(attributes_of_payload, key_mapping)
+    def from_request(payload_attributes)
+      rename_attributes(payload_attributes, from_request_key_mapping)
     end
 
     def to_response(model)
-      rename_attributes(model.attributes, inverted_key_mapping).as_json
+      rename_attributes(model.attributes, to_response_key_mapping).as_json
     end
 
     def rename_attributes(attributes, mapping)
-      mapping = mapping.with_indifferent_access
-      attributes
-        .to_hash
-        .except(*mapping.values)
-        .transform_keys! { |key| mapping[key] || key }
-        .with_indifferent_access
+      replace_keys(attributes.to_hash, mapping).with_indifferent_access
     end
 
-    def key_mapping
+    def replace_keys(hsh, mapping)
+      mapping.each do |k, v|
+        hsh[v] = hsh.delete(k)
+      end
+      hsh
+    end
+
+    def from_request_key_mapping
       {
         "created_at" => "device_created_at",
         "updated_at" => "device_updated_at"
       }
     end
 
-    def inverted_key_mapping
-      key_mapping.invert
+    def to_response_key_mapping
+      from_request_key_mapping.invert
     end
   end
 end

spec/controllers/api/v3/appointments_controller_spec.rb

@@ -338,6 +338,8 @@ def create_record_list(n, options = {})
         expect(response_1_records.count).to eq 4
         expect(response_1_records.map(&:facility).to_set).to eq Set[request_facility]
 
+        reset_controller
+
         # GET request 2
         get :sync_to_user, params: {limit: 4, process_token: response_1_body["process_token"]}
         response_2_body = JSON(response.body)