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..c9a223a 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" @@ -81,14 +82,21 @@ 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) + reset! + ExitHook.enable + @configuration = configuration + end # Returns the global configuration object # # @return [Config] the global configuration def configuration - @configuration ||= Config.new + @configuration ||= Config.new.tap { ExitHook.enable } end # Configure Langfuse globally @@ -153,10 +161,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 +418,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..1454c6e --- /dev/null +++ b/lib/langfuse/exit_hook.rb @@ -0,0 +1,77 @@ +# 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! + mutex.synchronize do + return if @installed + + Kernel.at_exit { run } + @installed = true + @active = false + end + end + + # Enable the installed callback for the current SDK lifecycle. + # + # @return [void] + def enable + mutex.synchronize { @active = true } + end + + # Disable the callback after an explicit shutdown or reset. + # + # @return [void] + def disable + 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 + mutex.synchronize do + active = @active + @active = false + active + end + end + + def reset_after_fork + @mutex = Mutex.new + end + + def mutex + @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..aa6aaa8 --- /dev/null +++ b/spec/langfuse/exit_hook_spec.rb @@ -0,0 +1,57 @@ +# 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 the inherited hook mutex in a forked child" do + skip "fork is not available" unless Process.respond_to?(:fork) + + parent_mutex = described_class.send(:mutex) + _child_pid, status, child_state = capture_forked_state do + { mutex_replaced: !described_class.send(:mutex).equal?(parent_mutex) } + end + + expect(status).to be_success + 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 2a6b667..3c8d4da 100644 --- a/spec/langfuse_spec.rb +++ b/spec/langfuse_spec.rb @@ -27,6 +27,41 @@ config2 = described_class.configuration expect(config1).to eq(config2) end + + 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 + + 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 @@ -179,6 +214,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 +266,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