3 files changed, 106 insertions(+), 60 deletions(-)

A lib/blather_client.rb
M lib/command.rb
M sgx_jmp.rb

@@ 0,0 1,69 @@
+# frozen_string_literal: true
+
+class BlatherClient < Blather::Client
+	def handle_data(stanza)
+		EMPromise.resolve(nil).then {
+			with_sentry(stanza) do |scope|
+				Thread.current[:log] = ::LOG.child(
+					transaction: scope.transaction_name
+				)
+				super
+			rescue StandardError => e
+				handle_error(scope, stanza, e)
+			end
+		}.catch { |e| panic(e) }
+	end
+
+	def with_sentry(stanza)
+		Sentry.clone_hub_to_current_thread
+
+		Sentry.with_scope do |scope|
+			setup_scope(stanza, scope)
+			yield scope
+		ensure
+			scope.get_transaction&.then do |tx|
+				tx.set_status("ok") unless tx.status
+				tx.finish
+			end
+		end
+	end
+
+	def setup_scope(stanza, scope)
+		name = stanza.respond_to?(:node) ? stanza.node : stanza.name
+		scope.clear_breadcrumbs
+		scope.set_transaction_name(name)
+		scope.set_user(jid: stanza.from.stripped.to_s)
+
+		transaction = Sentry.start_transaction(
+			name: name,
+			op: "blather.handle_data"
+		)
+		scope.set_span(transaction) if transaction
+	end
+
+	def handle_error(scope, stanza, e)
+		log.error(
+			"Error raised during #{scope.transaction_name}: " \
+			"#{e.class}",
+			e
+		)
+		Sentry.capture_exception(e) unless e.is_a?(Sentry::Error)
+		scope.get_transaction&.set_status("internal_error")
+		return if e.respond_to?(:replied?) && e.replied?
+
+		write stanza.as_error("internal-server-error", :cancel)
+	end
+
+	def call_handler(handler, guards, stanza)
+		result = if guards.first.respond_to?(:to_str)
+			found = stanza.find(*guards)
+			handler.call(stanza, found) unless found.empty?
+		else
+			handler.call(stanza) unless guarded?(guards, stanza)
+		end
+
+		return result unless result.is_a?(Promise)
+
+		result.sync
+	end
+end

@@ 49,16 49,15 @@ class Command
 			@blather = blather
 			@format_error = format_error
 			@iq = iq
-			@log = LOG.child(node: iq.node)
+			@log = Thread.current[:log] || LOG
 		end
 
 		def execute
 			StatsD.increment("command", tags: ["node:#{iq.node}"])
 			EMPromise.resolve(nil).then {
 				Command.execution = self
-				sentry_hub
 				catch_after(EMPromise.resolve(yield self))
-			}.catch(&method(:panic))
+			}
 		end
 
 		def reply(stanza=nil)


@@ 85,24 84,9 @@ class Command
 			EMPromise.reject(FinalStanza.new(reply))
 		end
 
-		def sentry_hub
-			return @sentry_hub if @sentry_hub
-
-			# Stored on Fiber-local in 4.3.1 and earlier
-			# https://github.com/getsentry/sentry-ruby/issues/1495
-			@sentry_hub = Sentry.get_current_hub
-			raise "Sentry.init has not been called" unless @sentry_hub
-
-			@sentry_hub.push_scope
-			@sentry_hub.current_scope.clear_breadcrumbs
-			@sentry_hub.current_scope.set_transaction_name(@iq.node)
-			@sentry_hub.current_scope.set_user(jid: @iq.from.stripped.to_s)
-			@sentry_hub
-		end
-
 		def customer
 			@customer ||= @customer_repo.find_by_jid(@iq.from.stripped).then { |c|
-				sentry_hub.current_scope.set_user(
+				Sentry.set_user(
 					id: c.customer_id,
 					jid: @iq.from.stripped
 				)


@@ 120,23 104,18 @@ class Command
 			}.catch_only(Timeout) { nil }.catch_only(FinalStanza) { |e|
 				@blather << e.stanza
 			}.catch do |e|
-				log_error(e)
 				send_final_error(e)
+				EMPromise.reject(e)
 			end
 		end
 
 		def send_final_error(e)
-			finish(@format_error.call(e), type: :error).catch_only(FinalStanza) do |s|
-				@blather << s.stanza
+			def e.replied?
+				true
 			end
-		end
 
-		def log_error(e)
-			@log.error("Error raised during #{iq.node}: #{e.class}", e)
-			if e.is_a?(::Exception)
-				sentry_hub.capture_exception(e)
-			else
-				sentry_hub.capture_message(e.to_s)
+			finish(@format_error.call(e), type: :error).catch_only(FinalStanza) do |s|
+				@blather << s.stanza
 			end
 		end
 	end

@@ 29,6 29,10 @@ EM::Hiredis.logger = LOG
 StatsD.logger = LOG
 LOG.info "Starting"
 
+def log
+	Thread.current[:log] || LOG
+end
+
 Sentry.init do |config|
 	config.logger = LOG
 	config.breadcrumbs_logger = [:sentry_logger]


@@ 169,7 173,7 @@ end
 BRAINTREE = AsyncBraintree.new(**CONFIG[:braintree])
 
 def panic(e, hub=nil)
-	(Thread.current[:log] || LOG).fatal(
+	log.fatal(
 		"Error raised during event loop: #{e.class}",
 		e
 	)


@@ 183,6 187,19 @@ end
 
 EM.error_handler(&method(:panic))
 
+require_relative "lib/blather_client"
+@client = BlatherClient.new
+
+setup(
+	CONFIG[:component][:jid],
+	CONFIG[:component][:secret],
+	CONFIG[:server][:host],
+	CONFIG[:server][:port],
+	nil,
+	nil,
+	async: true
+)
+
 # Infer anything we might have been notified about while we were down
 def catchup_notify_low_balance(db)
 	db.query(<<~SQL).each do |c|


@@ 225,7 242,7 @@ def load_plans_to_db!
 end
 
 when_ready do
-	LOG.info "Ready"
+	log.info "Ready"
 	BLATHER = self
 	REDIS = EM::Hiredis.connect
 	TEL_SELECTIONS = TelSelections.new


@@ 251,16 268,6 @@ when_ready do
 	Web.run(LOG.child, *WEB_LISTEN)
 end
 
-setup(
-	CONFIG[:component][:jid],
-	CONFIG[:component][:secret],
-	CONFIG[:server][:host],
-	CONFIG[:server][:port],
-	nil,
-	nil,
-	async: true
-)
-
 message to: /\Aaccount@/, body: /./ do |m|
 	StatsD.increment("deprecated_account_bot")
 


@@ 279,12 286,11 @@ before(
 before nil, to: /\Acustomer_/, from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/ do |s|
 	StatsD.increment("stanza_customer")
 
-	sentry_hub = new_sentry_hub(s, name: "stanza_customer")
-	CustomerRepo.new(set_user: sentry_hub.current_scope.method(:set_user)).find(
+	Sentry.set_transaction_name("stanza_customer")
+	CustomerRepo.new(set_user: Sentry.method(:set_user)).find(
 		s.to.node.delete_prefix("customer_")
-	).then { |customer|
-		customer.stanza_to(s)
-	}.catch { |e| panic(e, sentry_hub) }
+	).then { |customer| customer.stanza_to(s) }
+
 	halt
 end
 


@@ 294,7 300,7 @@ message(
 	from: /(\A|@)#{CONFIG[:sgx]}(\/|\Z)/
 ) do |m|
 	StatsD.increment("inbound_group_text")
-	sentry_hub = new_sentry_hub(m, name: "inbound_group_text")
+	Sentry.set_transaction_name("inbound_group_text")
 
 	address = m.find("ns:addresses", ns: ADDRESSES_NS).first
 		&.find("ns:address", ns: ADDRESSES_NS)


@@ 310,7 316,7 @@ message(
 			BLATHER << m
 		}.catch_only(CustomerRepo::NotFound) { |e|
 			BLATHER << m.as_error("forbidden", :auth, e.message)
-		}.catch { |e| panic(e, sentry_hub) }
+		}
 end
 
 # Ignore groupchat messages


@@ 347,9 353,8 @@ class CustomerExpired < StandardError; end
 message do |m|
 	StatsD.increment("message")
 
-	sentry_hub = new_sentry_hub(m, name: "message")
 	today = Time.now.utc.to_date
-	CustomerRepo.new(set_user: sentry_hub.current_scope.method(:set_user))
+	CustomerRepo.new(set_user: Sentry.method(:set_user))
 		.find_by_jid(m.from.stripped).then { |customer|
 			next customer.stanza_from(m) unless billable_message(m)
 


@@ 372,13 377,7 @@ message do |m|
 			BLATHER << m.as_error("policy-violation", :wait, e.message)
 		}.catch_only(CustomerRepo::NotFound, CustomerExpired) { |e|
 			BLATHER << m.as_error("forbidden", :auth, e.message)
-		}.catch { |e| panic(e, sentry_hub) }
-end
-
-message :error? do |m|
-	StatsD.increment("message_error")
-
-	LOG.error "MESSAGE ERROR", stanza: m
+		}
 end
 
 IQ_MANAGER = SessionManager.new(self, :id)


@@ 428,13 427,12 @@ end
 disco_items node: "http://jabber.org/protocol/commands" do |iq|
 	StatsD.increment("command_list")
 
-	sentry_hub = new_sentry_hub(iq, name: iq.node)
 	reply = iq.reply
 	reply.node = "http://jabber.org/protocol/commands"
 
 	CustomerRepo.new(
 		sgx_repo: Bwmsgsv2Repo.new,
-		set_user: sentry_hub.current_scope.method(:set_user)
+		set_user: Sentry.method(:set_user)
 	).find_by_jid(
 		iq.from.stripped
 	).catch {


@@ 450,7 448,7 @@ disco_items node: "http://jabber.org/protocol/commands" do |iq|
 			)
 		}
 		self << reply
-	}.catch { |e| panic(e, sentry_hub) }
+	}
 end
 
 iq "/iq/ns:services", ns: "urn:xmpp:extdisco:2" do |iq|