Driver tests for full lib/pq connection + necessary fixes

So it turns out that while bearing some relation to `lib/pq` through
various pieces that sqlc pulls in, the `riverdatabasesql` driver didn't
actually work when used with `lib/pq`. When testing it, we'd open a
`sql.DB` pool _via_ Pgx, so Pgx was doing all the heavy lifting.

Here, add an alternate driver test strategy to test `riverdatabasesql`
with a `lib/pq` pool, thereby ensuring that the full range of operations
works even with no Pgx in the mix at all.

Things were broken, so we get those fixed up. There we two main
problems:

* As described in #545, `lib/pq` doesn't handle intervals correctly, and
  reads them in as ints, which is a big problem because they're in
  nanoseconds. We work around this by injecting intervals (like leader
  TTLs) an second floats instead, which really doesn't feel that much
  worse.

* The custom bits class for `unique_states` didn't work. To fix this I
  move to an alternate strategy of inserts `unique_states` as an
  integer, a technique that I've been using in SQLite because there's no
  `bits` type. I could make a strong argument that the use of an integer
  over `bits` would've been a better design from the get go anyway,
  because it makes all the bit-wise arithmetic much simpler as you use
  C-style conventions like `states & (1 << 3)` that are very widespread
  and well understood. Either way though, things work reasonably well if
  we insert as an integer, then transition to storage in bits.

Fixes #545 and #566. Supersedes #882.

Author: brandur

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Job kinds must comply to a format of `\A[\w][\w\-\[\]<>\/.·:+]+\z`, mainly in an attempt to eliminate commas and spaces to make format more predictable for an upcoming search UI. This check can be disabled for now using `Config.SkipJobKindValidation`, but this option will likely be removed in a future version of River. [PR #879](https://github.com/riverqueue/river/pull/879).
 
+### Fixed
+
+- The `riverdatabasesql` now fully supports raw connections through [`lib/pq`](https://github.com/lib/pq) rather than just `database/sql` through Pgx. We don't recommend the use of `lib/pq` as it's an unmaintained project, but this change should help with compatibility for older projects. [PR #883](https://github.com/riverqueue/river/pull/883).
+
 ## [0.21.0] - 2025-05-02
 
 ⚠️ Internal APIs used for communication between River and River Pro have changed. If using River Pro, make sure to update River and River Pro to latest at the same time to get compatible versions. River v0.21.0 is compatible with River Pro v0.13.0.

driver_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/stdlib"
+	"github.com/lib/pq"
 	"github.com/stretchr/testify/require"
 
 	"github.com/riverqueue/river/internal/rivercommon"
@@ -19,10 +20,41 @@ import (
 	"github.com/riverqueue/river/riverdriver/riverdatabasesql"
 	"github.com/riverqueue/river/riverdriver/riverpgxv5"
 	"github.com/riverqueue/river/rivershared/riversharedtest"
+	"github.com/riverqueue/river/rivershared/util/urlutil"
 	"github.com/riverqueue/river/rivertype"
 )
 
-func TestDriverDatabaseSQL(t *testing.T) {
+func TestDriverDatabaseSQLLibPQ(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	connector, err := pq.NewConnector(urlutil.DatabaseSQLCompatibleURL(riversharedtest.TestDatabaseURL()))
+	require.NoError(t, err)
+
+	stdPool := sql.OpenDB(connector)
+	t.Cleanup(func() { require.NoError(t, stdPool.Close()) })
+
+	driver := riverdatabasesql.New(stdPool)
+
+	riverdrivertest.Exercise(ctx, t,
+		func(ctx context.Context, t *testing.T) (riverdriver.Driver[*sql.Tx], string) {
+			t.Helper()
+
+			return driver, riverdbtest.TestSchema(ctx, t, driver, nil)
+		},
+		func(ctx context.Context, t *testing.T) riverdriver.Executor {
+			t.Helper()
+
+			tx := riverdbtest.TestTx(ctx, t, driver, nil)
+
+			// TODO(brandur): Set `search_path` path here when SQLite changes come in.
+
+			return riverdatabasesql.New(nil).UnwrapExecutor(tx)
+		})
+}
+
+func TestDriverDatabaseSQLPgx(t *testing.T) {
 	t.Parallel()
 
 	var (

internal/riverinternaltest/riverdrivertest/riverdrivertest.go

@@ -3,6 +3,7 @@ package riverdrivertest
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"slices"
 	"sort"
@@ -2708,7 +2709,8 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 				defer close(goroutineDone)
 
 				_, err := otherExec.PGAdvisoryXactLock(ctx, 123456)
-				require.ErrorIs(t, err, context.Canceled)
+				// pgx will produce a context.Canceled error, but pg swallows it to emit its own
+				require.Regexp(t, "(context canceled|pq: canceling statement due to user request)", err.Error())
 			}()
 
 			select {
@@ -3387,22 +3389,30 @@ func exerciseListener[TTx any](ctx context.Context, t *testing.T, driverWithPool
 func requireEqualTime(t *testing.T, expected, actual time.Time) {
 	t.Helper()
 
-	// Leaving off the nanosecond portion has the effect of truncating it rather
-	// than rounding to the nearest microsecond, which functionally matches
-	// pgx's behavior while persisting.
 	const rfc3339Micro = "2006-01-02T15:04:05.999999Z07:00"
 
-	require.Equal(t,
-		expected.Format(rfc3339Micro),
-		actual.Format(rfc3339Micro),
+	// This is a bit unfortunate, but while Pgx truncates to the nearest
+	// microsecond, lib/pq will round, thereby producing off-by-one microsecond
+	// problems in tests without intervention. Here, allow either the truncated
+	// or rounded version. It's a bit gnarly, but the upcoming SQLite change
+	// brings in a better way to accomplish this, so it should be short-lived.
+	var (
+		actualFormatted            = actual.Format(rfc3339Micro)
+		expectedRoundedFormatted   = expected.Round(1 * time.Microsecond).Format(rfc3339Micro)
+		expectedTruncatedFormatted = expected.Format(rfc3339Micro)
 	)
+	require.True(t, expectedRoundedFormatted == actualFormatted || expectedTruncatedFormatted == actualFormatted,
+		"Expected time %s to be equal to either %s (rounded, for lib/pq) or %s (truncated, for Pgx)", actualFormatted, expectedRoundedFormatted, expectedTruncatedFormatted)
 }
 
 func requireMissingRelation(t *testing.T, err error, missingRelation string) {
 	t.Helper()
 
 	var pgErr *pgconn.PgError
-	require.ErrorAs(t, err, &pgErr)
-	require.Equal(t, pgerrcode.UndefinedTable, pgErr.Code)
-	require.Equal(t, fmt.Sprintf(`relation "%s" does not exist`, missingRelation), pgErr.Message)
+	if errors.As(err, &pgErr) {
+		require.Equal(t, pgerrcode.UndefinedTable, pgErr.Code)
+		require.Equal(t, fmt.Sprintf(`relation "%s" does not exist`, missingRelation), pgErr.Message)
+	} else {
+		require.ErrorContains(t, err, fmt.Sprintf("pq: relation %q does not exist", missingRelation))
+	}
 }

riverdbtest/riverdbtest.go

@@ -399,6 +399,11 @@ func TestTx[TTx any](ctx context.Context, tb testing.TB, driver riverdriver.Driv
 			return
 		}
 
+		// Cancelled context again, but this one from libpq.
+		if err.Error() == "driver: bad connection" {
+			return
+		}
+
 		// Similar to the above, but a newly appeared error that wraps the
 		// above. As far as I can tell, no error variables are available to use
 		// with `errors.Is`.

riverdriver/riverdatabasesql/internal/dbsqlc/models.go

@@ -49,6 +49,15 @@ sql:
           - db_type: "jsonb"
             go_type: "string"
 
+          - db_type: "pg_catalog.bit"
+            go_type: int
+
+          - db_type: "pg_catalog.bit"
+            go_type:
+              type: int
+              pointer: true
+            nullable: true
+
           - db_type: "pg_catalog.interval"
             go_type: "time.Duration"
 
@@ -60,23 +69,3 @@ sql:
               type: "time.Time"
               pointer: true
             nullable: true
-
-          # There doesn't appear to be a good type that's suitable for database/sql other
-          # than the ones in pgtype. It's not great to make the database/sql driver take
-          # a dependency on pgx, but the reality is most users will (or should) be using
-          # pgx anyway.
-          #
-          # Unfortunately due to some sqlc limitations, you can't just use the
-          # pgtype package directly (it tries to use the non-v5 import path and
-          # you end up with duplicate pgtype imports). So there's an alias
-          # package that exposes it indirectly.
-          - db_type: "pg_catalog.bit"
-            go_type:
-              import: "github.com/riverqueue/river/riverdriver/riverdatabasesql/internal/pgtypealias"
-              type: "Bits"
-
-          - db_type: "pg_catalog.bit"
-            go_type:
-              import: "github.com/riverqueue/river/riverdriver/riverdatabasesql/internal/pgtypealias"
-              type: "Bits"
-            nullable: true