Quiet per-user logs and stop app data update collisions

app/controllers/api/v8/users_controller.rb

@@ -392,10 +392,16 @@ def set_extra_data(user, eager_save = false)
           return if extra_fields.nil?
           namespace = extra_fields['namespace']
           raise 'Namespace not defined' unless namespace
-          extra_fields['data'].each do |key, value|
-            datum = user.user_app_data.find_or_initialize_by(namespace: namespace, field_name: key)
-            datum.value = value
-            datum.save! if eager_save
+          data = extra_fields['data']
+          # Some clients encode an empty map as [].
+          return if data.blank?
+          if eager_save
+            UserAppDatum.upsert_values(user_id: user.id, namespace: namespace, values_by_field_name: data.to_unsafe_h)
+          else
+            data.each do |key, value|
+              datum = user.user_app_data.find_or_initialize_by(namespace: namespace, field_name: key)
+              datum.value = value
+            end
           end
         end
 

app/models/user_app_datum.rb

@@ -2,5 +2,18 @@
 # frozen_string_literal: true
 
 class UserAppDatum < ApplicationRecord
+  extend T::Sig
+
   belongs_to :user
+
+  # Writes a persisted user's fields in one namespace atomically, so concurrent writes of the same field can't race to insert it.
+  sig { params(user_id: Integer, namespace: String, values_by_field_name: T::Hash[String, T.untyped]).void }
+  def self.upsert_values(user_id:, namespace:, values_by_field_name:)
+    return if values_by_field_name.empty?
+    # Sorted so concurrent upserts lock the rows in the same order and can't deadlock.
+    rows = values_by_field_name.sort.map do |field_name, value|
+      { user_id: user_id, namespace: namespace, field_name: field_name, value: value }
+    end
+    upsert_all(rows, unique_by: %i[user_id field_name namespace])
+  end
 end

config/environments/production.rb

@@ -58,8 +58,8 @@
   # Force all access to the app over SSL, use Strict-Transport-Security, and use secure cookies.
   # config.force_ssl = true
 
-  # Set to :debug to see everything in the log.
-  config.log_level = :info
+  # Set RAILS_LOG_LEVEL=debug to see everything in the log.
+  config.log_level = ENV.fetch('RAILS_LOG_LEVEL', 'info')
 
   # Prepend all log lines with the following tags.
   # config.log_tags = [ :subdomain, :uuid ]

lib/kafka_updater.rb

@@ -29,7 +29,7 @@ def update_user_progress(task)
       return true
     end
     course = Course.find_by!(id: task.course_id)
-    Rails.logger.info("Publishing progress for user #{user.id} with moocfi id: #{course.moocfi_id}.")
+    Rails.logger.debug { "Publishing progress for user #{user.id} with moocfi id: #{course.moocfi_id}." }
     if !course.moocfi_id || course.moocfi_id.blank?
       Rails.logger.error('Cannot publish progress because moocfi id is not specified')
       return finished_successfully
@@ -39,7 +39,7 @@ def update_user_progress(task)
     available_points = AvailablePoint.course_sheet_points(course, parts)
     user_points = points_per_user[user.username]
     unless user_points
-      Rails.logger.info('No points found. Skipping')
+      Rails.logger.debug('No points found. Skipping')
       return true
     end
     progress = user_points.map do |group_name, awarded_points|
@@ -62,7 +62,7 @@ def update_user_progress(task)
     }
     topic = task.realtime ? 'user-course-progress-realtime' : 'user-course-progress-batch'
     make_kafka_request(topic, message)
-    Rails.logger.info("Publishing progress finished for user #{user.id}")
+    Rails.logger.debug { "Publishing progress finished for user #{user.id}" }
     finished_successfully = true
     finished_successfully
   end
@@ -86,7 +86,7 @@ def update_course_progress(task)
         Rails.logger.warn("KafkaUpdater#update_course_progress: no user for login=#{username.inspect} course_id=#{course.id}")
         next
       end
-      Rails.logger.info("Publishing progress for user #{current_user.id}")
+      Rails.logger.debug { "Publishing progress for user #{current_user.id}" }
       progress = points_by_group.map do |group_name, awarded_points|
         max_points = available_points[group_name] || 0
         stupid_name = "osa#{group_name.tr('^0-9', '').rjust(2, "0")}"
@@ -106,7 +106,7 @@ def update_course_progress(task)
         message_format_version: 1
       }
       make_kafka_request('user-course-progress-batch', message)
-      Rails.logger.info("Publishing progress finished for user #{current_user.id}")
+      Rails.logger.debug { "Publishing progress finished for user #{current_user.id}" }
     end
     Rails.logger.info("Batch publishing progress finished for course #{course.name}")
     finished_successfully = true
@@ -122,7 +122,7 @@ def update_user_points(task)
       return true
     end
     course = Course.find_by!(id: task.course_id)
-    Rails.logger.info("Publishing points for user #{user.id} with moocfi id: #{course.moocfi_id}.")
+    Rails.logger.debug { "Publishing points for user #{user.id} with moocfi id: #{course.moocfi_id}." }
     if !course.moocfi_id
       Rails.logger.error('Cannot publish points because moocfi id is not specified. Removing extra task (this task never should have been created.)')
       task.destroy!
@@ -145,7 +145,7 @@ def update_user_points(task)
     }
     topic = task.realtime ? 'user-points-realtime' : 'user-points-batch'
     make_kafka_request(topic, message)
-    Rails.logger.info("Publishing points finished for user #{user.id}")
+    Rails.logger.debug { "Publishing points finished for user #{user.id}" }
     finished_successfully = true
     finished_successfully
   end
@@ -169,7 +169,7 @@ def update_course_points(task)
         Rails.logger.warn("KafkaUpdater#update_course_points: no user for login=#{username.inspect} course_id=#{course.id}")
         next
       end
-      Rails.logger.info("Publishing points for user #{current_user.id}")
+      Rails.logger.debug { "Publishing points for user #{current_user.id}" }
       exercises_data = []
       exercises.map do |exercise|
         awarded_points = exercise.points_for(current_user)
@@ -200,7 +200,7 @@ def update_course_points(task)
         message_format_version: 1
       }
       make_kafka_request('user-course-points-batch', message)
-      Rails.logger.info("Publishing points finished for user #{current_user.id}")
+      Rails.logger.debug { "Publishing points finished for user #{current_user.id}" }
     end
     Rails.logger.info("Batch publishing points finished for course #{course.name}")
     finished_successfully = true
@@ -254,7 +254,7 @@ def update_user_course_points(task)
       return true
     end
     course = Course.find_by!(id: task.course_id)
-    Rails.logger.info("Publishing points for user #{user.id} with moocfi id: #{course.moocfi_id}.")
+    Rails.logger.debug { "Publishing points for user #{user.id} with moocfi id: #{course.moocfi_id}." }
     if !course.moocfi_id
       Rails.logger.error('Cannot publish points because moocfi id is not specified')
       return finished_successfully
@@ -290,7 +290,7 @@ def update_user_course_points(task)
     }
     topic = task.realtime ? 'user-course-points-realtime' : 'user-course-points-batch'
     make_kafka_request(topic, message)
-    Rails.logger.info("Publishing points finished for user #{user.id}")
+    Rails.logger.debug { "Publishing points finished for user #{user.id}" }
     finished_successfully = true
     finished_successfully
   end

lib/student_submission_migrator.rb

@@ -100,14 +100,8 @@ def update_app_data
       return unless migration_entry['update_app_data']
       app_data_updates = migration_entry['update_app_data']
       app_data_updates.each do |adu|
-        namespace = adu['namespace']
-        adu['set_values'].each do |record|
-          field_name = record['field_name']
-          value = record['value']
-          db_record = UserAppDatum.find_or_initialize_by(namespace: namespace, field_name: field_name, user: @user)
-          db_record.value = value
-          db_record.save!
-        end
+        values_by_field_name = adu['set_values'].to_h { |record| [record['field_name'], record['value']] }
+        UserAppDatum.upsert_values(user_id: @user.id, namespace: adu['namespace'], values_by_field_name: values_by_field_name)
       end
     end
 end

script/background_daemon

@@ -19,20 +19,18 @@ def exception_to_str(e)
   e.message + "\n" + e.backtrace.join("\n        from ")
 end
 
-Dir.chdir root_dir
-require root_dir + '/config/environment'
-
-stdout_logger = Logger.new(STDOUT)
-stdout_logger.formatter = proc do |severity, datetime, _progname, msg|
-  "#{datetime}: #{severity} - #{msg}\n"
+def build_logger(target)
+  logger = Logger.new(target, level: Rails.application.config.log_level)
+  logger.formatter = proc do |severity, datetime, _progname, msg|
+    "#{datetime}: #{severity} - #{msg}\n"
+  end
+  logger
 end
 
-file_logger = Logger.new(Rails.root.join('log/background_daemon.log'))
-file_logger.formatter = proc do |severity, datetime, _progname, msg|
-  "#{datetime}: #{severity} - #{msg}\n"
-end
+Dir.chdir root_dir
+require root_dir + '/config/environment'
 
-Rails.logger = ActiveSupport::BroadcastLogger.new(stdout_logger, file_logger)
+Rails.logger = ActiveSupport::BroadcastLogger.new(build_logger(STDOUT), build_logger(Rails.root.join('log/background_daemon.log')))
 FileUtils.mkdir_p(Rails.root.join('log/background_daemon'))
 
 Rails.logger.auto_flushing = true if Rails.logger.respond_to? :auto_flushing=
@@ -45,10 +43,7 @@ processes = BackgroundTaskRegistry.all_tasks.map do |task|
   Process.fork do
     # Create individual logger for this task
     task_log_file = Rails.root.join('log', 'background_daemon', "#{task.class.name.underscore}.log")
-    task_logger = Logger.new(task_log_file)
-    task_logger.formatter = proc do |severity, datetime, _progname, msg|
-      "#{datetime}: #{severity} - #{msg}\n"
-    end
+    task_logger = build_logger(task_log_file)
     task_logger.auto_flushing = true if task_logger.respond_to? :auto_flushing=
 
     # Set up Rails logger for this process to use the task-specific logger

spec/controllers/api/v8/users_controller_spec.rb

@@ -179,6 +179,51 @@ def do_update(old_password)
     end
   end
 
+  describe 'PUT update extra fields' do
+    let!(:token) { Doorkeeper::AccessToken.create!(resource_owner_id: user.id, scopes: 'public') }
+
+    def do_update(data, **request_options)
+      put :update, params: { id: 'current', user: { email: user.email, extra_fields: { namespace: 'ns', data: data } } }, **request_options
+    end
+
+    def app_data_values
+      UserAppDatum.where(user: user, namespace: 'ns').pluck(:field_name, :value).to_h
+    end
+
+    it 'updates existing fields and adds new ones' do
+      existing = UserAppDatum.create!(user: user, namespace: 'ns', field_name: 'a', value: 'old', created_at: 1.day.ago, updated_at: 1.day.ago)
+
+      do_update({ a: 'new', b: 'added' })
+
+      expect(response).to have_http_status(200)
+      expect(app_data_values).to eq('a' => 'new', 'b' => 'added')
+      expect(existing.reload.created_at).to be_within(1.second).of(1.day.ago)
+      expect(existing.updated_at).to be_within(1.minute).of(Time.current)
+    end
+
+    it 'overwrites a field that a concurrent request created after this one started' do
+      competitor_inserted = false
+      allow(UserAppDatum).to receive(:upsert_all).and_wrap_original do |original, *args, **kwargs|
+        competitor_inserted = true
+        UserAppDatum.insert!({ user_id: user.id, namespace: 'ns', field_name: 'a', value: 'concurrent' })
+        original.call(*args, **kwargs)
+      end
+
+      do_update({ a: 'mine' })
+
+      expect(competitor_inserted).to be(true)
+      expect(response).to have_http_status(200)
+      expect(app_data_values).to eq('a' => 'mine')
+    end
+
+    it 'accepts an empty data list' do
+      do_update([], as: :json)
+
+      expect(response).to have_http_status(200)
+      expect(app_data_values).to be_empty
+    end
+  end
+
   describe 'DELETE destroy' do
     before :each do
       controller.current_user = admin

spec/lib/student_submission_migrator_spec.rb

@@ -66,6 +66,18 @@
     expect { expect(migrator.migration_is_allowed).to be(true) }.not_to output.to_stdout
   end
 
+  it 'sets the configured app data' do
+    UserAppDatum.create!(user: @user, namespace: 'ns', field_name: 'a', value: 'old')
+    allow(SiteSetting).to receive(:value).and_return([{
+      'from' => @course.id, 'to' => @other_course.id,
+      'update_app_data' => [{ 'namespace' => 'ns', 'set_values' => [{ 'field_name' => 'a', 'value' => 'new' }, { 'field_name' => 'b', 'value' => 'added' }] }]
+    }])
+
+    StudentSubmissionMigrator.new(@course, @other_course, @user).migrate!
+
+    expect(UserAppDatum.where(user: @user, namespace: 'ns').pluck(:field_name, :value).to_h).to eq('a' => 'new', 'b' => 'added')
+  end
+
   it 'throws exception if it cannot be migrated safely' do
     bad_course = FactoryBot.create(:course)
     allow(bad_course).to receive(:git_revision).and_return('other')