diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md index 375c49ab8..91c982f1f 100644 --- a/ruby/CHANGELOG.md +++ b/ruby/CHANGELOG.md @@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Default periodic jobs to their occurrence time, preserving constructor and argument-level schedules. [PR #1537](https://github.com/riverqueue/river/pull/1537). + - Fail retries that omit a saved cursor-only step, while preserving their checkpoints for recovery. Jobs with an existing step marker retain their behavior, including legacy stale cursor entries. [PR #1534](https://github.com/riverqueue/river/pull/1534). - Support large SQLite metadata updates without reaching the expression-depth limit, preserving top-level replacement and JSON null values. [PR #1533](https://github.com/riverqueue/river/pull/1533). diff --git a/ruby/driver/riverqueue-activerecord/lib/driver.rb b/ruby/driver/riverqueue-activerecord/lib/driver.rb index 9694e374e..9363a608c 100644 --- a/ruby/driver/riverqueue-activerecord/lib/driver.rb +++ b/ruby/driver/riverqueue-activerecord/lib/driver.rb @@ -164,7 +164,7 @@ def transaction(&) private_constant :SQLITE_CONFLICT_WHERE, :SQLITE_JOB_COLUMNS, :UNIQUE_INSERT_METADATA_KEY private def format_time(time) - time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N") + normalize_time(time).strftime("%Y-%m-%d %H:%M:%S.%3N") end private def notification_query(sql) diff --git a/ruby/driver/riverqueue-sequel/lib/driver.rb b/ruby/driver/riverqueue-sequel/lib/driver.rb index b8b619ccd..895022800 100644 --- a/ruby/driver/riverqueue-sequel/lib/driver.rb +++ b/ruby/driver/riverqueue-sequel/lib/driver.rb @@ -155,7 +155,7 @@ def transaction private_constant :SQLITE_CONFLICT_WHERE, :SQLITE_JOB_COLUMNS, :UNIQUE_INSERT_METADATA_KEY private def format_time(time) - time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N") + normalize_time(time).strftime("%Y-%m-%d %H:%M:%S.%3N") end private def postgres_insert_params_to_hash(insert_params, nonce) diff --git a/ruby/lib/client.rb b/ruby/lib/client.rb index 478a4461b..27b61123e 100644 --- a/ruby/lib/client.rb +++ b/ruby/lib/client.rb @@ -49,6 +49,13 @@ def __finish_claimed_job(row, error = nil) @runtime.finish_claimed(row, error) end + # Internal scheduler entry point. Prepare ordinary insertion options and + # uniqueness before applying an occurrence fallback to the stored schedule. + def __insert_with_default_scheduled_at(args, insert_opts, default_scheduled_at) + insert_params = make_insert_params(args, insert_opts, default_scheduled_at) + run_insert_plugins([insert_params]) { [insert_and_check_unique_job(insert_params)] }.first + end + def __perform_job(id, allow_scheduled: false) @runtime.perform_job(id, allow_scheduled: allow_scheduled) end @@ -107,8 +114,7 @@ def id = config.id def insert(args, insert_opts: nil, **options) raise ArgumentError, "use insert_opts or keyword options, not both" if insert_opts && !options.empty? insert_opts ||= options.empty? ? EMPTY_INSERT_OPTS : InsertOpts.new(**options) - insert_params = make_insert_params(args, insert_opts) - run_insert_plugins([insert_params]) { [insert_and_check_unique_job(insert_params)] }.first + __insert_with_default_scheduled_at(args, insert_opts, nil) end # Inserts many new jobs as part of a single batch operation for improved @@ -425,7 +431,7 @@ def subscribe(*kinds, buffer_size: 100) JobInsertResult.new(job, unique_skipped_as_duplicated: unique_skipped_as_duplicate) end - private def make_insert_params(args, insert_opts) + private def make_insert_params(args, insert_opts, default_scheduled_at = nil) raise ArgumentError, "args should respond to `#kind`" if !args.respond_to?(:kind) # ~all objects in Ruby respond to `#to_json`, so check non-nil instead. @@ -470,6 +476,9 @@ def subscribe(*kinds, buffer_size: 100) insert_params.unique_states = unique_states end + if scheduled_at.nil? && default_scheduled_at + insert_params.scheduled_at = default_scheduled_at.getutc + end insert_params end diff --git a/ruby/lib/client_runtime.rb b/ruby/lib/client_runtime.rb index ac064ca8e..47cb2acaf 100644 --- a/ruby/lib/client_runtime.rb +++ b/ruby/lib/client_runtime.rb @@ -838,10 +838,10 @@ def wake end private def run_periodic(now) - due = @periodic_jobs.due(now) do |job, error| + due = @periodic_jobs.__due_with_times(now) do |job, error| @config.logger.error("River periodic job schedule failed (id=#{job.id.inspect}): #{error.full_message}") end - due.each do |periodic_job| + due.each do |periodic_job, scheduled_at| value = periodic_job.constructor.call next unless value @@ -849,7 +849,7 @@ def wake opts = (opts || InsertOpts.new).dup opts.metadata = (opts.metadata || {}).merge("periodic" => true) opts.metadata["river:periodic_job_id"] = periodic_job.id if periodic_job.id && !periodic_job.id.empty? - @client.insert(args, insert_opts: opts) + @client.__insert_with_default_scheduled_at(args, opts, scheduled_at) rescue => error @config.logger.error("River periodic job failed to insert: #{error.full_message}") end diff --git a/ruby/lib/driver/runtime.rb b/ruby/lib/driver/runtime.rb index 7325aa657..6a7a8a7c4 100644 --- a/ruby/lib/driver/runtime.rb +++ b/ruby/lib/driver/runtime.rb @@ -5,12 +5,19 @@ module River::Driver # Database operations used by River's worker runtime. Driver gems implement a # small set of raw SQL primitives and include this module so the state machine - # stays identical across ActiveRecord and Sequel. + # stays identical across ActiveRecord and Sequel. Those private primitives + # provide execution, quoting, row access, and connection management inside + # driver modules; callers use the driver's public operations and Ruby values. module Runtime def init_driver postgres_capabilities if runtime_postgres? end + # Returns the UTC timestamp that writing and reading +time+ preserves. + def normalize_time(time) + runtime_postgres? ? time.getutc.floor(6) : time.getutc.round(3) + end + def postgres_capabilities @postgres_capabilities_cache.fetch(runtime_connection_pool) do row = runtime_query_rows(<<~SQL).first @@ -471,6 +478,11 @@ def request_resign runtime_query_rows(sql) end + # Smallest timestamp increment preserved by this driver, in seconds. + def time_precision + runtime_postgres? ? Rational(1, 1_000_000) : Rational(1, 1_000) + end + private def runtime_append_error(error) value = error.respond_to?(:to_h) ? error.to_h : error if runtime_postgres? @@ -782,9 +794,9 @@ def request_resign cast = runtime_postgres? ? "::timestamptz" : "" encoded = if runtime_postgres? - value.getutc.iso8601(6) + normalize_time(value).iso8601(6) else - value.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N") + normalize_time(value).strftime("%Y-%m-%d %H:%M:%S.%3N") end "#{runtime_quote(encoded)}#{cast}" end diff --git a/ruby/lib/periodic_job.rb b/ruby/lib/periodic_job.rb index 85aad7970..453d9bbdf 100644 --- a/ruby/lib/periodic_job.rb +++ b/ruby/lib/periodic_job.rb @@ -76,7 +76,10 @@ def add_many(jobs) raise ArgumentError, "periodic jobs require leader election" if !@enabled && !jobs.empty? now = Time.now.utc - entries = jobs.map { |job| {job: job, next_at: job.run_on_start ? now : job.next_at(now)} } + entries = jobs.map do |job| + run_on_start = job.run_on_start + {job: job, next_at: run_on_start ? now : job.next_at(now), run_on_start: run_on_start} + end handles = @mutex.synchronize do ids = @jobs.values.to_h { |entry| [entry[:job].id, true] } jobs.each do |job| @@ -100,7 +103,9 @@ def clear self end - def due(now) + # Internal scheduler entry point that retains each claimed occurrence's + # time even after advancing its registration. + def __due_with_times(now) # Application schedules may reenter the registry, so evaluate outside its # lock and only advance registrations that haven't been removed or claimed. entries = @mutex.synchronize { @jobs.map { |handle, entry| [handle, entry, entry[:next_at]] } } @@ -110,8 +115,10 @@ def due(now) following = entry[:job].next_at(now) @mutex.synchronize do next unless @jobs[handle].equal?(entry) && entry[:next_at] == next_at + scheduled_at = entry[:run_on_start] ? now : next_at + entry[:run_on_start] = false entry[:next_at] = following - entry[:job] + [entry[:job], scheduled_at] end rescue => error raise unless block_given? @@ -120,6 +127,10 @@ def due(now) end end + def due(now, &block) + __due_with_times(now, &block).map(&:first) + end + # Returns true when the bundle has no registered periodic jobs. def empty? @mutex.synchronize { @jobs.empty? } diff --git a/ruby/sig/client.rbs b/ruby/sig/client.rbs index ca9eaea25..699972e55 100644 --- a/ruby/sig/client.rbs +++ b/ruby/sig/client.rbs @@ -15,6 +15,7 @@ module River def initialize: (_Driver driver, ?config: Config?) -> void def __finish_claimed_job: (JobRow, ?Exception?) -> untyped + def __insert_with_default_scheduled_at: (jobArgs, InsertOpts, Time?) -> JobInsertResult def __perform_job: (Integer, ?allow_scheduled: bool) -> untyped def __interrupt_workers: () -> untyped def __runtime_healthy?: () -> bool @@ -62,7 +63,7 @@ module River TAG_RE: Regexp private def insert_and_check_unique_job: (Driver::JobInsertParams) -> JobInsertResult - private def make_insert_params: (jobArgs, InsertOpts) -> Driver::JobInsertParams + private def make_insert_params: (jobArgs, InsertOpts, ?Time?) -> Driver::JobInsertParams private def make_unique_key_and_bitmask: (Driver::JobInsertParams, UniqueOpts) -> [String, String] private def run_insert_plugins: (Array[Driver::JobInsertParams]) { () -> Array[JobInsertResult] } -> Array[JobInsertResult] private def truncate_time: (Time, Integer | Rational) -> Time diff --git a/ruby/sig/driver.rbs b/ruby/sig/driver.rbs index 10a1eb41d..bcc0b3760 100644 --- a/ruby/sig/driver.rbs +++ b/ruby/sig/driver.rbs @@ -28,6 +28,7 @@ module River def leader_acquire: (String, ?ttl: Integer, ?now: Time) -> bool def leader_release: (String) -> untyped def leader_renew: (String, ?ttl: Integer, ?now: Time) -> bool + def normalize_time: (Time) -> Time def notification_delete_before: (horizon: Time, ?max: Integer) -> Integer def notification_listener: () -> untyped def request_resign: () -> nil @@ -37,6 +38,7 @@ module River def queue_resume: (String, ?now: Time) -> Array[Queue] def queue_update: (String, metadata: Hash[untyped, untyped], ?now: Time) -> Queue? def queue_upsert: (String, ?metadata: Hash[untyped, untyped], ?now: Time) -> Queue? + def time_precision: () -> Rational def transaction: [T] () { () -> T } -> T # this set of methods is used only in tests @@ -126,6 +128,7 @@ module River def runtime_decode_postgres_job: (untyped) -> JobRow @postgres_capabilities_cache: PostgresCapabilities::Cache def init_driver: () -> void + def normalize_time: (Time) -> Time def postgres_capabilities: () -> PostgresCapabilities def postgres_notify_insert: (Array[JobInsertParams]) -> void def job_get_cancelled_ids: (Array[Integer]) -> Array[Integer] @@ -158,6 +161,8 @@ module River def queue_update: (untyped, metadata: untyped, ?now: untyped) -> untyped def queue_upsert: (untyped, ?metadata: untyped, ?now: untyped) -> untyped + def time_precision: () -> Rational + private def job_get_by_id: (untyped) -> untyped diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs index 0610ab634..2782a3465 100644 --- a/ruby/sig/runtime.rbs +++ b/ruby/sig/runtime.rbs @@ -337,6 +337,7 @@ module River @next_handle: Integer @wake: untyped def initialize: (Array[PeriodicJob], wake: untyped, ?enabled: bool) -> void + def __due_with_times: (Time) ?{ (PeriodicJob, StandardError) -> void } -> Array[[PeriodicJob, Time]] def add: (PeriodicJob) -> Integer def add_many: (Array[PeriodicJob]) -> Array[Integer] def clear: () -> self diff --git a/ruby/spec/client_runtime_branch_spec.rb b/ruby/spec/client_runtime_branch_spec.rb index 999be512c..5510cad55 100644 --- a/ruby/spec/client_runtime_branch_spec.rb +++ b/ruby/spec/client_runtime_branch_spec.rb @@ -625,6 +625,24 @@ def execute(runtime, value) expect(output.string).to include("River maintenance service failed", "persistent service failure") end + it "passes the scheduled occurrence to periodic insertion even when a pass is late" do + occurrence = nil + periodic = River::PeriodicJob.new(schedule: ->(time) { + following = time + 60 + occurrence ||= following + following + }) { :args } + observed = [] + client = Object.new + client.define_singleton_method(:insert) { |_args, insert_opts:| observed << insert_opts.scheduled_at } + client.define_singleton_method(:__insert_with_default_scheduled_at) { |_args, _opts, fallback| observed << fallback } + value = described_class.new(client, Object.new, config.with(periodic_jobs: [periodic])) + + value.send(:run_periodic, occurrence + 3_600) + + expect(observed).to eq([occurrence]) + end + it "logs and isolates a periodic constructor failure" do output = StringIO.new periodic = River::PeriodicJob.new( @@ -645,7 +663,7 @@ def execute(runtime, value) good = River::PeriodicJob.new(schedule: River::PeriodicInterval.new(60), run_on_start: true) { :args } inserted = [] client = Object.new - client.define_singleton_method(:insert) { |args, **| inserted << args } + client.define_singleton_method(:__insert_with_default_scheduled_at) { |args, _options, _scheduled_at| inserted << args } value = described_class.new(client, Object.new, config.with(logger: Logger.new(output), periodic_jobs: [broken, good])) value.send(:run_periodic, Time.now.utc + 1) diff --git a/ruby/spec/client_spec.rb b/ruby/spec/client_spec.rb index d9fe3a346..2a8dfe37f 100644 --- a/ruby/spec/client_spec.rb +++ b/ruby/spec/client_spec.rb @@ -55,6 +55,63 @@ def to_json = JSON.dump({order_id: order_id, customer_id: customer_id, trace_id: let!(:driver) { River::Driver::Sequel.new(available_test_database) } let(:client) { River::Client.new(driver) } + describe "#__insert_with_default_scheduled_at" do + [ + ["occurrence", nil, nil, 9, 12], + ["argument", 13, nil, 13, 13], + ["constructor", 13, 14, 14, 14] + ].each do |source, argument_hour, constructor_hour, expected_hour, period_hour| + it "preserves the #{source} schedule and prepared uniqueness key" do + now = Time.utc(2026, 1, 1, 12, 30) + fallback = Time.utc(2026, 1, 1, 9, 15) + argument_schedule = Time.utc(2026, 1, 1, argument_hour, 15) if argument_hour + constructor_schedule = Time.utc(2026, 1, 1, constructor_hour, 15) if constructor_hour + argument_options = River::InsertOpts.new(scheduled_at: argument_schedule).freeze + options = River::InsertOpts.new(scheduled_at: constructor_schedule, + unique_opts: River::UniqueOpts.new(by_period: 3_600)).freeze + reads = 0 + args = SimpleArgs.new(job_num: 1).tap do |value| + value.define_singleton_method(:insert_opts) do + reads += 1 + argument_options + end + value.freeze + end + observed = nil + plugin = Object.new + plugin.define_singleton_method(:insert_begin) { |params| observed = [params.args, params.scheduled_at, params.unique_key] } + client = River::Client.new(driver, config: River::Config.new(plugins: [plugin])) + client.instance_variable_set(:@time_now_utc, -> { now }) + + row = client.__insert_with_default_scheduled_at(args, options, fallback).job + + expected_schedule = Time.utc(2026, 1, 1, expected_hour, 15) + expected_key = Digest::SHA256.digest("&kind=simple&period=2026-01-01T#{format("%02d", period_hour)}:00:00Z") + expect(row.scheduled_at).to eq(expected_schedule) + expect(row.unique_key).to eq(expected_key) + expect(observed).to eq([args, expected_schedule, expected_key]) + expect(observed.first).to equal(args) + expect(reads).to eq(1) + expect(argument_options.scheduled_at).to eq(argument_schedule) + expect(options.scheduled_at).to eq(constructor_schedule) + end + end + + it "keeps the prepared state and uniqueness key with a future fallback" do + now = Time.utc(2026, 1, 1, 12, 30) + fallback = Time.utc(2026, 1, 1, 14, 15) + options = River::InsertOpts.new(unique_opts: River::UniqueOpts.new(by_period: 3_600)) + client.instance_variable_set(:@time_now_utc, -> { now }) + + row = client.__insert_with_default_scheduled_at(SimpleArgs.new(job_num: 1), options, fallback).job + + expected_key = Digest::SHA256.digest("&kind=simple&period=2026-01-01T12:00:00Z") + expect(row.scheduled_at).to eq(fallback) + expect(row.state).to eq(River::JOB_STATE_AVAILABLE) + expect(row.unique_key).to eq(expected_key) + end + end + describe "#insert" do it "normalizes symbolic kinds and argument-level options before calling plugins and drivers" do args = SimpleArgsWithInsertOpts.new(job_num: 1) diff --git a/ruby/spec/driver_runtime_feature_spec.rb b/ruby/spec/driver_runtime_feature_spec.rb index 3c793561f..ceaa61ac1 100644 --- a/ruby/spec/driver_runtime_feature_spec.rb +++ b/ruby/spec/driver_runtime_feature_spec.rb @@ -13,6 +13,14 @@ it_behaves_like "driver queue and leadership state" end +RSpec.describe "SQLite driver timestamp contract", database: :sqlite do + around(:each) { |example| sqlite_test_transaction(&example) } + + let(:driver) { River::Driver::Sequel.new(SQLITE_DB) } + + it_behaves_like "driver timestamp precision" +end + RSpec.describe "SQLite numeric metadata counters", database: :sqlite do around(:each) { |example| sqlite_test_transaction(&example) } diff --git a/ruby/spec/driver_runtime_shared_examples.rb b/ruby/spec/driver_runtime_shared_examples.rb index 058f54ce4..b0a74d6ac 100644 --- a/ruby/spec/driver_runtime_shared_examples.rb +++ b/ruby/spec/driver_runtime_shared_examples.rb @@ -1,5 +1,7 @@ # frozen_string_literal: true +require "delegate" + class DriverRuntimeArgs def initialize(value) @value = value @@ -10,6 +12,55 @@ def kind = "driver_runtime" def to_json = JSON.dump(value: @value) end +RSpec.shared_examples "driver timestamp precision" do + let(:client) { River::Client.new(driver) } + + describe "timestamp precision" do + it "exposes timestamp operations through a delegating driver" do + delegated = SimpleDelegator.new(driver) + time = Time.utc(2026, 10, 25) + Rational(123_500_001, 1_000_000_000) + + expect(delegated.normalize_time(time)).to eq(driver.normalize_time(time)) + expect(delegated.time_precision).to eq(driver.time_precision) + end + + [ + ["a timestamp just below the millisecond rounding boundary", 123_499_999, Rational(123_499, 1_000_000), Rational(123, 1_000)], + ["a timestamp just above the millisecond rounding boundary", 123_500_001, Rational(123_500, 1_000_000), Rational(124, 1_000)], + ["a timestamp just before a second rollover", 999_999_999, Rational(999_999, 1_000_000), 1] + ].each do |name, nanoseconds, postgres_fraction, sqlite_fraction| + it "normalizes #{name} to its persisted UTC value" do + midnight = Time.utc(2026, 10, 25) + expected_fraction = (driver.migration_backend == :sqlite) ? sqlite_fraction : postgres_fraction + time = (midnight + Rational(nanoseconds, 1_000_000_000)).getlocal("+05:30") + expected = midnight + expected_fraction + normalized = driver.normalize_time(time) + driver.queue_upsert("precision", now: time) + + expect(normalized).to eq(expected) + expect(normalized).to be_utc + expect(driver.normalize_time(normalized)).to eq(normalized) + expect(driver.queue_get("precision").created_at).to eq(expected) + inserted = client.insert(DriverRuntimeArgs.new(1), scheduled_at: time).job + expect(client.job_get(inserted.id).scheduled_at).to eq(expected) + end + end + + it "preserves one timestamp increment when written and read back" do + expected = (driver.migration_backend == :sqlite) ? Rational(1, 1_000) : Rational(1, 1_000_000) + expect(driver.time_precision).to be_a(Rational) + expect(driver.time_precision).to eq(expected) + time = driver.normalize_time(Time.utc(2026, 10, 25) + Rational(123_500_001, 1_000_000_000)) + driver.queue_upsert("precision", now: time) + driver.queue_update("precision", metadata: {}, now: time + driver.time_precision) + + queue = driver.queue_get("precision") + expect(queue.updated_at.to_r - queue.created_at.to_r).to eq(driver.time_precision) + expect(queue.updated_at).to be > queue.created_at + end + end +end + RSpec.shared_examples "overflowing metadata counters" do let(:client) { River::Client.new(driver) } let(:now) { Time.utc(2026, 1, 2, 3, 4, 5) } @@ -43,6 +94,7 @@ def to_json = JSON.dump(value: @value) end RSpec.shared_examples "driver job state machine" do + it_behaves_like "driver timestamp precision" it_behaves_like "overflowing metadata counters" let(:client) { River::Client.new(driver) } diff --git a/ruby/spec/periodic_job_spec.rb b/ruby/spec/periodic_job_spec.rb index 98d82312e..5b0796f9f 100644 --- a/ruby/spec/periodic_job_spec.rb +++ b/ruby/spec/periodic_job_spec.rb @@ -189,6 +189,23 @@ def periodic(id: nil, run_on_start: false, schedule: ->(time) { time + 60 }) expect(jobs.due(Time.now.utc + 1)).to eq([job]) end + it "timestamps delayed startup work once and keeps later occurrence times" do + calls = [] + job = periodic(run_on_start: true, schedule: ->(time) { + calls << time + time + 60 + }) + jobs = described_class.new([job], wake: wake) + first_pass = Time.now.utc + 3_600 + expect(calls).to be_empty + + expect(jobs.__due_with_times(first_pass)).to eq([[job, first_pass]]) + expect(calls).to eq([first_pass]) + expect(jobs.__due_with_times(first_pass + 59)).to be_empty + expect(jobs.__due_with_times(first_pass + 65)).to eq([[job, first_pass + 60]]) + expect(calls).to eq([first_pass, first_pass + 65]) + end + it "does not return future jobs" do jobs = described_class.new([periodic], wake: wake)