[ruby] preserve periodic occurrence schedules

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).

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)

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)

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
 

ruby/lib/client_runtime.rb

@@ -838,18 +838,18 @@ 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
 
         args, opts = value.is_a?(Array) ? value : [value, nil]
         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

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

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? }

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

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

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

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)

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)

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) }