From 708101bf89d2a8b6ef31e5a9e1cc608b6693a057 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ianar=C3=A9=20S=C3=A9vi?= Date: Mon, 9 Nov 2026 18:31:20 +0100 Subject: [PATCH] :recycle: various improvements to polling --- .rubocop.yml | 11 ++ lib/mindee/v1/client.rb | 1 - lib/mindee/v1/parsing/common/execution.rb | 4 - lib/mindee/v2/client.rb | 131 ++++++++++++++---- .../v2/parsing/field/inference_fields.rb | 4 - lib/mindee/v2/parsing/job.rb | 2 - sig/mindee/v2/client.rbs | 21 ++- spec/v2/client_v2_spec.rb | 2 +- 8 files changed, 135 insertions(+), 41 deletions(-) diff --git a/.rubocop.yml b/.rubocop.yml index 14601ccc9..d57c0509f 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -37,11 +37,22 @@ Metrics/MethodLength: Metrics/ClassLength: Max: 200 + Metrics/ModuleLength: Max: 200 + Metrics/ParameterLists: Max: 8 +Metrics/CyclomaticComplexity: + Max: 10 + +Metrics/PerceivedComplexity: + Max: 10 + +Style/OptionalBooleanParameter: + Enabled: false + Metrics/AbcSize: Max: 60 diff --git a/lib/mindee/v1/client.rb b/lib/mindee/v1/client.rb index 1a5de1405..1852e9dbb 100644 --- a/lib/mindee/v1/client.rb +++ b/lib/mindee/v1/client.rb @@ -232,7 +232,6 @@ def parse_queued(job_id, product_class, endpoint: nil) # @param endpoint [Mindee::V1::HTTP::Endpoint] Endpoint of the API. # @param cancellation_token [Mindee::HTTP::CancellationToken, nil] Token for cancellation. # @return [Mindee::V1::Parsing::Common::ApiResponse] - # rubocop:disable-next Metrics/PerceivedComplexity, Metrics/CyclomaticComplexity def enqueue_and_parse( input_source, product_class, diff --git a/lib/mindee/v1/parsing/common/execution.rb b/lib/mindee/v1/parsing/common/execution.rb index bfb9f1a24..b1a266c1f 100644 --- a/lib/mindee/v1/parsing/common/execution.rb +++ b/lib/mindee/v1/parsing/common/execution.rb @@ -46,8 +46,6 @@ class Execution # @return [String] attr_reader :workflow_id - # rubocop:disable Metrics/CyclomaticComplexity - # @param product_class [Mindee::Inference] # @param http_response [Hash] def initialize(product_class, http_response) @@ -69,8 +67,6 @@ def initialize(product_class, http_response) @uploaded_at = Time.iso8601(http_response['uploaded_at']) if http_response['uploaded_at'] @workflow_id = http_response['workflow_id'] end - - # rubocop:enable Metrics/CyclomaticComplexity end end end diff --git a/lib/mindee/v2/client.rb b/lib/mindee/v2/client.rb index c75cb468f..7643b8c9b 100644 --- a/lib/mindee/v2/client.rb +++ b/lib/mindee/v2/client.rb @@ -84,7 +84,6 @@ def get_job(job_id) # @param polling_options [Hash, PollingOptions, nil] Parameters for polling. # @param cancellation_token [Mindee::HTTP::CancellationToken, nil] Token for cancellation. # @return [Parsing::BaseResponse] - # rubocop:disable-next Metrics/CyclomaticComplexity def enqueue_and_get_result( product, input_source, @@ -122,7 +121,13 @@ def enqueue_and_get_result( raise Mindee::Error::MindeeError, 'Enqueueing of the document was canceled.' if cancellation_token&.canceled? - poll_on_job(product, enqueue_response, polling_options_instance, cancellation_token) + poll_for_product_result( + product, + enqueue_response, + polling_options_instance, + false, + cancellation_token + ) end # Search for resources matching the given criteria. @@ -143,53 +148,125 @@ def search_models(model_name, model_type) private + # Checks if all webhooks associated with a job have finished processing. + # @param job_response [Mindee::V2::Parsing::JobResponse] + # @return [Boolean] + def check_webhooks_done?(job_response) + are_webhooks_done = job_response.job.webhooks.nil? || + job_response.job.webhooks.all? { |w| w.status != 'Processing' } + + if are_webhooks_done + logger.debug('All webhooks are completed.') + return true + end + + logger.debug('Not all webhooks are completed.') + false + end + + # Wait and retry logic for polling operations. + # @param try_counter [Integer] Current retry attempt index. + # @param max_retries [Integer] Maximum number of retries. + # @param polling_options [Input::PollingOptions] Options for polling. + # @param cancellation_token [Mindee::HTTP::CancellationToken, nil] Token for cancellation. + def throttle_polling_attempt( + try_counter, + max_retries, + polling_options, + cancellation_token = nil + ) + if try_counter >= max_retries + logger.debug('Not waiting, max retries reached.') + return + end + + raise Mindee::Error::MindeeError, 'Polling was canceled.' if cancellation_token&.canceled? + + logger.debug("Waiting #{polling_options.delay_sec} seconds before next attempt...") + sleep(polling_options.delay_sec) + end + # Polls a job until it is processed or the maximum number of tries is reached. - # @param product [Class] - # @param initial_response [V2::Parsing::JobResponse] + # @param initial_response [Mindee::V2::Parsing::JobResponse] # @param polling_options [Input::PollingOptions] # @param cancellation_token [Mindee::HTTP::CancellationToken, nil] Token for cancellation. - # @return [Parsing::BaseResponse] + # @return [Parsing::JobResponse] def poll_on_job( - product, initial_response, polling_options, - cancellation_token + wait_for_webhooks = false, + cancellation_token = nil ) + logger.debug( + "Waiting #{polling_options.initial_delay_sec} seconds before attempting to retrieve the result..." + ) sleep(polling_options.initial_delay_sec) - try_counter = 1 + + try_counter = 0 max_retries = polling_options.max_retries polling_url = initial_response.job.polling_url - job_response = get_job_by_url(polling_url) - while try_counter < max_retries - if job_response.job.status == 'Failed' - break - elsif !job_response.job.result_url.nil? - return get_result_by_url(product, job_response.job.result_url) - end + raise Mindee::Error::MindeeError, 'Polling was canceled.' if cancellation_token&.canceled? logger.debug( - "Poll attempt #{try_counter + 1} of #{max_retries}." + "Poll attempt #{try_counter + 1} of #{max_retries}" ) - raise Mindee::Error::MindeeError, 'Enqueueing of the document was canceled.' if cancellation_token&.canceled? - - sleep(polling_options.delay_sec) job_response = get_job_by_url(polling_url) + + if job_response.job.status == 'Processed' + logger.debug( + "Job ID #{job_response.job.id} completed processing at: #{job_response.job.completed_at}" + ) + + return job_response if !wait_for_webhooks || check_webhooks_done?(job_response) + end + + if job_response.job.status == 'Failed' + if job_response.job.error + raise Mindee::Error::MindeeHTTPErrorV2.new(job_response.job.error), + "Parsing failed for job #{job_response.job.id}" + end + + raise Mindee::Error::MindeeError, + "Parsing failed for job #{job_response.job.id}: No error detail available." + end + try_counter += 1 + throttle_polling_attempt(try_counter, max_retries, polling_options, cancellation_token) end - error = job_response.job.error - unless error.nil? - err_to_raise = Mindee::Error::MindeeHTTPErrorV2.new(error) - # NOTE: purposefully decoupled from the line above, otherwise rubocop thinks `error` is a `message` param. - raise err_to_raise + raise Mindee::Error::MindeeError, "Couldn't retrieve the result after #{try_counter} tries." + end + + # Poll until the inference results are retrieved or the max number of attempts is reached. + # @param product [Class] + # @param enqueue_response [V2::Parsing::JobResponse] + # @param polling_options [Input::PollingOptions] + # @param wait_for_webhooks [Boolean] + # @param cancellation_token [Mindee::HTTP::CancellationToken, nil] + # @return [Mindee::V2::Parsing::BaseResponse] + def poll_for_product_result( + product, + enqueue_response, + polling_options, + wait_for_webhooks = false, + cancellation_token = nil + ) + job_response = poll_on_job( + enqueue_response, + polling_options, + wait_for_webhooks, + cancellation_token + ) + + if job_response.job.result_url.nil? || job_response.job.result_url.strip.empty? + raise Mindee::Error::MindeeError, + 'The result URL is undefined. This is a server error, try again later or contact support.' end - sec_count = polling_options.delay_sec * try_counter - raise Mindee::Error::MindeeError, - "Asynchronous parsing request timed out after #{sec_count} seconds" + get_result_by_url(product, job_response.job.result_url) end end end diff --git a/lib/mindee/v2/parsing/field/inference_fields.rb b/lib/mindee/v2/parsing/field/inference_fields.rb index e7d0431f3..1b7738abf 100644 --- a/lib/mindee/v2/parsing/field/inference_fields.rb +++ b/lib/mindee/v2/parsing/field/inference_fields.rb @@ -63,8 +63,6 @@ def get_object_field(key) field end - # rubocop:disable Metrics/CyclomaticComplexity - # rubocop:disable Metrics/PerceivedComplexity # Convert the fields to a string representation. # @param indent [Integer, nil] Optional indentation level. # @return [String] String representation of all fields. @@ -96,8 +94,6 @@ def to_s(indent = 0) lines.join("\n").rstrip end - # rubocop:enable Metrics/CyclomaticComplexity - # rubocop:enable Metrics/PerceivedComplexity end end end diff --git a/lib/mindee/v2/parsing/job.rb b/lib/mindee/v2/parsing/job.rb index e3f726d8e..ad975af78 100644 --- a/lib/mindee/v2/parsing/job.rb +++ b/lib/mindee/v2/parsing/job.rb @@ -32,7 +32,6 @@ class Job # @return [ErrorResponse, nil] Error details when the job failed. attr_reader :error - # rubocop:disable Metrics/CyclomaticComplexity # @param server_response [Hash] Parsed JSON payload from the API. def initialize(server_response) raise ArgumentError, 'server_response must be a Hash' unless server_response.is_a?(Hash) @@ -56,7 +55,6 @@ def initialize(server_response) @webhooks.push JobWebhook.new(webhook) end end - # rubocop:enable Metrics/CyclomaticComplexity # String representation. # @return [String] diff --git a/sig/mindee/v2/client.rbs b/sig/mindee/v2/client.rbs index 23a89ee9c..a82ccacc6 100644 --- a/sig/mindee/v2/client.rbs +++ b/sig/mindee/v2/client.rbs @@ -42,12 +42,29 @@ module Mindee private - def poll_on_job: [T] ( + def check_webhooks_done?: (V2::Parsing::JobResponse job_response) -> bool + + def poll_for_product_result: [T] ( HTTP::_ProductClass[T] product, + V2::Parsing::JobResponse enqueu_response, + Input::PollingOptions polling_options, + ?bool wait_for_webhooks, + ?(Mindee::HTTP::CancellationToken | nil) cancellation_token + ) -> T + + def poll_on_job: ( V2::Parsing::JobResponse initial_response, Input::PollingOptions polling_options, + ?bool wait_for_webhooks, + ?(Mindee::HTTP::CancellationToken | nil) cancellation_token + ) -> V2::Parsing::JobResponse + + def throttle_polling_attempt: ( + Integer try_counter, + Integer max_retries, + Input::PollingOptions polling_options, Mindee::HTTP::CancellationToken | nil cancellation_token - ) -> T + ) -> void end end end diff --git a/spec/v2/client_v2_spec.rb b/spec/v2/client_v2_spec.rb index 24a144cdb..af7619383 100644 --- a/spec/v2/client_v2_spec.rb +++ b/spec/v2/client_v2_spec.rb @@ -200,7 +200,7 @@ def stub_next_request_with(method, hash:, status_code: 0) client.enqueue_and_get_result( Mindee::V2::Product::Extraction::Extraction, input_doc, params ) - end.to raise_error(Mindee::Error::MindeeError, %r{timed out}) + end.to raise_error(Mindee::Error::MindeeError, %r{Couldn't retrieve}) end end