Fix custom schema support of rivertest.Worker

[zmwangx]

Currently, rivertest.Worker ignores custom schema in river config.

The fix is one line. Also added a test case for it.

Simple testcontainers-based repro:

package repro

import (
	"context"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/riverqueue/river"
	"github.com/riverqueue/river/riverdriver/riverpgxv5"
	"github.com/riverqueue/river/rivermigrate"
	"github.com/riverqueue/river/rivertest"
	"github.com/stretchr/testify/require"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/postgres"
	"github.com/testcontainers/testcontainers-go/wait"
)

type Args struct{}

func (Args) Kind() string { return "repro" }

// TestRivertestWorkerCustomSchema reproduces the rivertest.Worker custom-schema bug.
//
//	PRE-FIX  (released river through v0.38.0): FAILS with
//	  "failed to update job to running state: ... relation \"river_job\" does not exist"
//	POST-FIX (replace -> patched fork): PASSES.
//
// River is migrated into a NON-default schema, and the job is worked through a
// transaction whose search_path doesn't include that schema. The insert (via
// the client) qualifies river_job with the schema, but the pre-fix running-state
// update runs unqualified and can't find the table.
func TestRivertestWorkerCustomSchema(t *testing.T) {
	ctx := context.Background()

	pgc, err := postgres.Run(ctx, "postgres:18-alpine",
		postgres.WithDatabase("river"),
		postgres.WithUsername("postgres"),
		postgres.WithPassword("postgres"),
		testcontainers.WithWaitStrategy(
			wait.ForLog("database system is ready to accept connections").
				WithOccurrence(2).WithStartupTimeout(60*time.Second),
		),
	)
	require.NoError(t, err)
	t.Cleanup(func() { _ = pgc.Terminate(ctx) })

	dsn, err := pgc.ConnectionString(ctx, "sslmode=disable")
	require.NoError(t, err)

	pool, err := pgxpool.New(ctx, dsn)
	require.NoError(t, err)
	t.Cleanup(pool.Close)

	const schema = "custom_schema"

	_, err = pool.Exec(ctx, "CREATE SCHEMA "+schema)
	require.NoError(t, err)
	migrator, err := rivermigrate.New(riverpgxv5.New(pool), &rivermigrate.Config{Schema: schema})
	require.NoError(t, err)
	_, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, nil)
	require.NoError(t, err)

	config := &river.Config{ID: "repro", Schema: schema}
	tx, err := pool.Begin(ctx)
	require.NoError(t, err)
	t.Cleanup(func() { _ = tx.Rollback(ctx) })

	worker := river.WorkFunc(func(ctx context.Context, job *river.Job[Args]) error { return nil })
	tw := rivertest.NewWorker(t, riverpgxv5.New(pool), config, worker)

	res, err := tw.Work(ctx, t, tx, Args{}, nil)
	require.NoError(t, err) // <-- fails pre-fix
	require.Equal(t, river.EventKindJobCompleted, res.EventKind)
}

Disclosure: Claude Opus assisted in authoring the fix. Everything's carefully human reviewed, and PR is entirely human authored.

[brandur]

Great, thank you!!

A couple minor nits below. Could you also add a changelog entry under the "fixed" section please?

[brandur]

The tests are currently ordered as:

• TestWorker_Work
• TestWorker_WorkJob

With subtests under each for different things. It seems a bit odd to pull out this new responsibility into a separate top-level test and put Work + WorkJob under it.

Can you ask Claude to reorder so that "CustomSchema" becomes a subtest under each of TestWorker_Work and TestWorker_WorkJob instead?

[zmwangx]

I greenlighted a separate test because these need a different setup, but yeah, probably better to fold in with inline setup instead. Done.

[brandur]

thx.

[brandur]

@zmwangx Thanks! And can you add that changelog too?

[zmwangx]

Oh sorry, missed that comment. Pushed.