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