From 7a2564f2bd11fde6087e8b106312885d89dee729 Mon Sep 17 00:00:00 2001 From: kadekillary Date: Sun, 16 Aug 2026 23:08:15 -0700 Subject: [PATCH 1/3] fix(runtime): flush pending telemetry at process exit --- docs/CONFIGURATION.md | 8 ++++ lib/langfuse.rb | 8 +++- lib/langfuse/exit_hook.rb | 82 +++++++++++++++++++++++++++++++++ spec/langfuse/exit_hook_spec.rb | 62 +++++++++++++++++++++++++ spec/langfuse_spec.rb | 13 ++++++ 5 files changed, 171 insertions(+), 2 deletions(-) create mode 100644 lib/langfuse/exit_hook.rb create mode 100644 spec/langfuse/exit_hook_spec.rb diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index f8c7fd9..aa74341 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -712,6 +712,14 @@ The SDK restores its background state after `fork` on Ruby 3.2 and later. This a The parent keeps its original queue and worker objects. Shutdown remains independent in each process. +## Process Exit + +The SDK automatically flushes pending spans and scores during normal process exit. One-off scripts and short-lived tasks do not need to register an `at_exit` callback. + +Call `Langfuse.shutdown` when the application must flush before process exit. An explicit shutdown disables the pending exit callback. `Langfuse.reset!` also disables the callback until the next SDK configuration lifecycle begins. + +The exit callback runs once and does not allow shutdown errors to escape the process-exit path. Abrupt termination, such as `SIGKILL`, cannot run process-exit callbacks. + ## Configuration by Environment ### Development diff --git a/lib/langfuse.rb b/lib/langfuse.rb index 85ba6e4..8d1aa9b 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -41,6 +41,7 @@ class UnauthorizedError < ApiError; end require_relative "langfuse/config" require_relative "langfuse/fork_safety" +require_relative "langfuse/exit_hook" require_relative "langfuse/cache_constants" require_relative "langfuse/prompt_cache" require_relative "langfuse/prompt_fetch_result" @@ -88,6 +89,7 @@ class << self # # @return [Config] the global configuration def configuration + ExitHook.enable @configuration ||= Config.new end @@ -153,10 +155,11 @@ def tracer_provider # @param timeout [Integer] Timeout in seconds # @return [void] # - # @example In a Rails initializer or shutdown hook - # at_exit { Langfuse.shutdown } + # @example Explicit early shutdown + # Langfuse.shutdown # def shutdown(timeout: 30) + ExitHook.disable client.shutdown if @client OtelSetup.shutdown(timeout: timeout) end @@ -409,6 +412,7 @@ def create_trace_id(seed: nil) # # @return [void] def reset! + ExitHook.disable client.shutdown if @client OtelSetup.shutdown(timeout: 5) if OtelSetup.initialized? @configuration = nil diff --git a/lib/langfuse/exit_hook.rb b/lib/langfuse/exit_hook.rb new file mode 100644 index 0000000..c7374f1 --- /dev/null +++ b/lib/langfuse/exit_hook.rb @@ -0,0 +1,82 @@ +# frozen_string_literal: true + +require_relative "fork_safety" + +module Langfuse + # Flushes SDK telemetry during normal process exit. + # + # @api private + module ExitHook + class << self + # Install the process callback once and enable it. + # + # @return [void] + def install! + install_mutex.synchronize do + return if @installed + + Kernel.at_exit { run } + @installed = true + @active = true + end + end + + # Enable the installed callback for the current SDK lifecycle. + # + # @return [void] + def enable + state_mutex.synchronize { @active = true } + end + + # Disable the callback after an explicit shutdown or reset. + # + # @return [void] + def disable + state_mutex.synchronize { @active = false } + end + + # Run the callback once without allowing shutdown errors to escape. + # + # @return [void] + def run + return unless consume_active_hook + + Langfuse.shutdown + rescue StandardError => e + warn_failure(e) + end + + private + + def consume_active_hook + state_mutex.synchronize do + active = @active + @active = false + active + end + end + + def reset_after_fork + @install_mutex = Mutex.new + @state_mutex = Mutex.new + end + + def install_mutex + @install_mutex ||= Mutex.new + end + + def state_mutex + @state_mutex ||= Mutex.new + end + + def warn_failure(error) + Kernel.warn("Langfuse exit flush failed: #{error.class} - #{error.message}") + rescue StandardError + nil + end + end + + install! + ForkSafety.register(self) + end +end diff --git a/spec/langfuse/exit_hook_spec.rb b/spec/langfuse/exit_hook_spec.rb new file mode 100644 index 0000000..c7a4ccb --- /dev/null +++ b/spec/langfuse/exit_hook_spec.rb @@ -0,0 +1,62 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Langfuse::ExitHook do + after do + described_class.enable + end + + describe ".install!" do + it "registers one process callback" do + installed = described_class.instance_variable_get(:@installed) + described_class.instance_variable_set(:@installed, false) + allow(Kernel).to receive(:at_exit) + + described_class.install! + described_class.install! + + expect(Kernel).to have_received(:at_exit).once + ensure + described_class.instance_variable_set(:@installed, installed) + end + end + + describe ".run" do + it "runs shutdown once" do + described_class.enable + expect(Langfuse).to receive(:shutdown).once + + described_class.run + described_class.run + end + + it "warns without raising when shutdown fails" do + described_class.enable + allow(Langfuse).to receive(:shutdown).and_raise(Langfuse::ApiError, "unavailable") + expect(Kernel).to receive(:warn).with(/Langfuse exit flush failed: Langfuse::ApiError - unavailable/) + + expect { described_class.run }.not_to raise_error + end + end + + describe "fork safety" do + it "replaces inherited hook mutexes in a forked child" do + skip "fork is not available" unless Process.respond_to?(:fork) + + parent_install_mutex = described_class.send(:install_mutex) + parent_state_mutex = described_class.send(:state_mutex) + _child_pid, status, child_state = capture_forked_state do + { + install_mutex_replaced: !described_class.send(:install_mutex).equal?(parent_install_mutex), + state_mutex_replaced: !described_class.send(:state_mutex).equal?(parent_state_mutex) + } + end + + expect(status).to be_success + expect(child_state).to eq(install_mutex_replaced: true, state_mutex_replaced: true) + expect(described_class.send(:install_mutex)).to equal(parent_install_mutex) + expect(described_class.send(:state_mutex)).to equal(parent_state_mutex) + end + end +end diff --git a/spec/langfuse_spec.rb b/spec/langfuse_spec.rb index 2a6b667..f79a573 100644 --- a/spec/langfuse_spec.rb +++ b/spec/langfuse_spec.rb @@ -27,6 +27,12 @@ config2 = described_class.configuration expect(config1).to eq(config2) end + + it "enables the process-exit callback" do + expect(Langfuse::ExitHook).to receive(:enable).and_call_original + + described_class.configuration + end end describe ".configure" do @@ -179,6 +185,12 @@ end describe ".reset!" do + it "disables the process-exit callback" do + expect(Langfuse::ExitHook).to receive(:disable).and_call_original + + described_class.reset! + end + it "resets configuration and client" do described_class.configure { |c| c.public_key = "test" } described_class.reset! @@ -225,6 +237,7 @@ end it "calls OtelSetup.shutdown with timeout" do + expect(Langfuse::ExitHook).to receive(:disable).and_call_original expect(Langfuse::OtelSetup).to receive(:shutdown).with(timeout: 30) described_class.shutdown end From 29cf976f82ffe4f74df3596b19e7afdf469ca2dd Mon Sep 17 00:00:00 2001 From: kadekillary Date: Mon, 17 Aug 2026 07:40:02 -0700 Subject: [PATCH 2/3] fix(shutdown): preserve disabled exit hook --- lib/langfuse.rb | 11 ++++++++--- lib/langfuse/exit_hook.rb | 21 ++++++++------------- spec/langfuse/exit_hook_spec.rb | 15 +++++---------- spec/langfuse_spec.rb | 11 ++++++++++- 4 files changed, 31 insertions(+), 27 deletions(-) diff --git a/lib/langfuse.rb b/lib/langfuse.rb index 8d1aa9b..da88f9a 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -82,15 +82,20 @@ class UnauthorizedError < ApiError; end module Langfuse # rubocop:disable Metrics/ClassLength class << self + # Set the global configuration object and start its process-exit lifecycle. + # # @param configuration [Config] the global configuration object - attr_writer :configuration + # @return [Config] the assigned configuration + def configuration=(configuration) + ExitHook.enable + @configuration = configuration + end # Returns the global configuration object # # @return [Config] the global configuration def configuration - ExitHook.enable - @configuration ||= Config.new + @configuration ||= Config.new.tap { ExitHook.enable } end # Configure Langfuse globally diff --git a/lib/langfuse/exit_hook.rb b/lib/langfuse/exit_hook.rb index c7374f1..1454c6e 100644 --- a/lib/langfuse/exit_hook.rb +++ b/lib/langfuse/exit_hook.rb @@ -12,12 +12,12 @@ class << self # # @return [void] def install! - install_mutex.synchronize do + mutex.synchronize do return if @installed Kernel.at_exit { run } @installed = true - @active = true + @active = false end end @@ -25,14 +25,14 @@ def install! # # @return [void] def enable - state_mutex.synchronize { @active = true } + mutex.synchronize { @active = true } end # Disable the callback after an explicit shutdown or reset. # # @return [void] def disable - state_mutex.synchronize { @active = false } + mutex.synchronize { @active = false } end # Run the callback once without allowing shutdown errors to escape. @@ -49,7 +49,7 @@ def run private def consume_active_hook - state_mutex.synchronize do + mutex.synchronize do active = @active @active = false active @@ -57,16 +57,11 @@ def consume_active_hook end def reset_after_fork - @install_mutex = Mutex.new - @state_mutex = Mutex.new - end - - def install_mutex - @install_mutex ||= Mutex.new + @mutex = Mutex.new end - def state_mutex - @state_mutex ||= Mutex.new + def mutex + @mutex ||= Mutex.new end def warn_failure(error) diff --git a/spec/langfuse/exit_hook_spec.rb b/spec/langfuse/exit_hook_spec.rb index c7a4ccb..aa6aaa8 100644 --- a/spec/langfuse/exit_hook_spec.rb +++ b/spec/langfuse/exit_hook_spec.rb @@ -41,22 +41,17 @@ end describe "fork safety" do - it "replaces inherited hook mutexes in a forked child" do + it "replaces the inherited hook mutex in a forked child" do skip "fork is not available" unless Process.respond_to?(:fork) - parent_install_mutex = described_class.send(:install_mutex) - parent_state_mutex = described_class.send(:state_mutex) + parent_mutex = described_class.send(:mutex) _child_pid, status, child_state = capture_forked_state do - { - install_mutex_replaced: !described_class.send(:install_mutex).equal?(parent_install_mutex), - state_mutex_replaced: !described_class.send(:state_mutex).equal?(parent_state_mutex) - } + { mutex_replaced: !described_class.send(:mutex).equal?(parent_mutex) } end expect(status).to be_success - expect(child_state).to eq(install_mutex_replaced: true, state_mutex_replaced: true) - expect(described_class.send(:install_mutex)).to equal(parent_install_mutex) - expect(described_class.send(:state_mutex)).to equal(parent_state_mutex) + expect(child_state).to eq(mutex_replaced: true) + expect(described_class.send(:mutex)).to equal(parent_mutex) end end end diff --git a/spec/langfuse_spec.rb b/spec/langfuse_spec.rb index f79a573..d9a13d7 100644 --- a/spec/langfuse_spec.rb +++ b/spec/langfuse_spec.rb @@ -28,11 +28,20 @@ expect(config1).to eq(config2) end - it "enables the process-exit callback" do + it "enables the process-exit callback when it builds a configuration" do + described_class.reset! expect(Langfuse::ExitHook).to receive(:enable).and_call_original described_class.configuration end + + it "does not re-enable the callback when it reads a cached configuration" do + configuration = described_class.configuration + Langfuse::ExitHook.disable + expect(Langfuse::ExitHook).not_to receive(:enable) + + expect(described_class.configuration).to equal(configuration) + end end describe ".configure" do From f380f4853d53cb32d87f4f8d297e5e2f4fb942f1 Mon Sep 17 00:00:00 2001 From: kadekillary Date: Mon, 17 Aug 2026 11:25:05 -0700 Subject: [PATCH 3/3] fix(runtime): reset client on configuration replacement --- lib/langfuse.rb | 1 + spec/langfuse_spec.rb | 20 ++++++++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/lib/langfuse.rb b/lib/langfuse.rb index da88f9a..c9a223a 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -87,6 +87,7 @@ class << self # @param configuration [Config] the global configuration object # @return [Config] the assigned configuration def configuration=(configuration) + reset! ExitHook.enable @configuration = configuration end diff --git a/spec/langfuse_spec.rb b/spec/langfuse_spec.rb index d9a13d7..3c8d4da 100644 --- a/spec/langfuse_spec.rb +++ b/spec/langfuse_spec.rb @@ -42,6 +42,26 @@ expect(described_class.configuration).to equal(configuration) end + + it "starts a new client lifecycle when configuration is replaced" do + previous_client = described_class.client + described_class.shutdown + replacement = Langfuse::Config.new do |config| + config.public_key = "replacement_pk" + config.secret_key = "replacement_sk" + end + request = stub_request(:post, "https://cloud.langfuse.com/api/public/ingestion") + .to_return(status: 200, body: "", headers: {}) + + described_class.configuration = replacement + replacement_client = described_class.client + replacement_client.create_score(name: "quality", value: 1) + replacement_client.flush_scores + + expect(replacement_client).not_to equal(previous_client) + expect(replacement_client.config).to equal(replacement) + expect(request).to have_been_requested.once + end end describe ".configure" do