From fd87a0ecbfb9038db805be9d2939a594718e4ae6 Mon Sep 17 00:00:00 2001 From: Blake Gentry Date: Sat, 24 Oct 2026 20:42:37 -0500 Subject: [PATCH] [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) }