From fd87a0ecbfb9038db805be9d2939a594718e4ae6 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sat, 24 Oct 2026 20:42:37 -0500 Subject: [PATCH 1/2] [ruby] expose driver timestamp semantics Driver consumers need to compare persisted timestamps without inspecting which database backend encodes them. Private SQL helpers also cannot be forwarded through delegating driver wrappers. Expose `normalize_time` and `time_precision` on the internal driver contract. Share normalization with SQL timestamp encoding and SQLite insertion formatting, preserving microsecond truncation on Postgres and millisecond rounding on SQLite. Return precision as an exact duration in seconds. Shared driver examples cover persisted queue and job timestamps at precision boundaries, UTC conversion, delegation, and the smallest representable increment across both adapters and database engines. --- .../riverqueue-activerecord/lib/driver.rb | 2 +- ruby/driver/riverqueue-sequel/lib/driver.rb | 2 +- ruby/lib/driver/runtime.rb | 18 +++++-- ruby/sig/driver.rbs | 5 ++ ruby/spec/driver_runtime_feature_spec.rb | 8 +++ ruby/spec/driver_runtime_shared_examples.rb | 52 +++++++++++++++++++ 6 files changed, 82 insertions(+), 5 deletions(-) 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/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/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/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) } From c2541e0ca5ff9a4e98156e7fa1c752fcfbfff11f Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sat, 24 Oct 2026 21:21:59 -0500 Subject: [PATCH 2/2] [ruby] preserve periodic occurrence schedules Delayed periodic work currently uses insertion time instead of the schedule's occurrence time. Share the ordinary insertion pipeline and retain one argument-option lookup. Preserve explicit constructor and argument schedules, along with ordinary key and state preparation, then apply an occurrence fallback to the stored schedule before insertion hooks run. Carry occurrence times from the registry without changing its existing due-job API. Timestamp initial startup work at its first pass and later work at its occurrence. Cover option precedence with frozen objects, prepared keys and states, late periodic passes, and delayed startup work. --- ruby/CHANGELOG.md | 2 + ruby/lib/client.rb | 15 +++++-- ruby/lib/client_runtime.rb | 6 +-- ruby/lib/periodic_job.rb | 17 ++++++-- ruby/sig/client.rbs | 3 +- ruby/sig/runtime.rbs | 1 + ruby/spec/client_runtime_branch_spec.rb | 20 ++++++++- ruby/spec/client_spec.rb | 57 +++++++++++++++++++++++++ ruby/spec/periodic_job_spec.rb | 17 ++++++++ 9 files changed, 127 insertions(+), 11 deletions(-) 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/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/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/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/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)