Let test workers call JobCompleteTx + test work transaction variants

[brandur]

Follows up #753 to make it possible to call JobCompleteTx inside test
worker implementations. The test worker tries to update an equivalent
job in the database's state to running so JobCompleteTx will be able
to find it when called (it only considers jobs in running state).

We also augment the API so that it's workable for users who want to use
test transactions in their tests (which is hopefully the vast majority
of users) by adding transaction variants WorkTx and WorkJobTx.

[brandur]

Ugh, lint auto correction.

[brandur]

@bgentry This is a little unfortunate — you still won't be able to use JobCompleteTx for the Work/WorkTx variants because there's no way to inject the ID of the job which should be updated.

There might be ways to work around that, but none of them are good. e.g. We could have WorkTx insert a job record as running, but then you're producing data into the database in people's tests, which probably isn't good (i.e. you probably wouldn't be able to do this for Work because you don't know whether they're in a test transaction or not).

[bgentry]

This is a little unfortunate — you still won't be able to use JobCompleteTx for the Work/WorkTx variants because there's no way to inject the ID of the job which should be updated.

Can you explain a little more what you mean here? Even if it's a synthetic job, JobCompleteTx would still get called—it's just that it will error because the job doesn't actually exist, right? Maybe that was a mistake and we should just go all in w/ inserting real job rows.

[brandur]

JobCompleteTx` would still get called—it's just that it will error because the job doesn't actually exist, right?

Yeah, exactly. There's no way for a user to put in a job row that'll let JobCompleteTx be called successfully.

Maybe that was a mistake and we should just go all in w/ inserting real job rows.

Yes, there's a good chance of that.

[bgentry]

Trying to think of what else should get updated here to be realistic. I guess it's all of these, and not just the state?

river/riverdriver/riverpgxv5/internal/dbsqlc/river_job.sql

Lines 151 to 154 in 0a8ee30

	state = 'running',
	attempt = river_job.attempt + 1,
	attempted_at = now(),
	attempted_by = array_append(river_job.attempted_by, @attempted_by::text)

So we should consider both setting those when updating the job, and also making sure they're set on the actual job struct.

[brandur]

Yeah, hmm. Definitely opens a can of worms. It'd be better to be able to offload a real insert to the client/executor so we don't have to muddle with all this.

[bgentry]

The docs at the top of the file about not touching the database are no longer accurate.

On a bigger note though, I'm wondering if we're thinking about this the right way. The original vision here was to not hit the DB which is why we're synthetically creating fake jobs instead of inserting and then working them. If we're going to start making DB queries in most cases regardless of whether it's a real job or a fake one, is that original design still reasonable?

I'm also a little worried about the potential to introduce test flakiness by attempting to update a job whose ID was synthetically generated from the in-memory sequence as it may correspond to a real (but different) job in the table.

Perhaps we should just go all in by always using transactions and always inserting real job records? We could even leverage the inline completer after execution. I think that might resolve any potential functionality issues I can think of—although we'd probably need to extract some of the executor logic to be able to reuse it here.

[bgentry]

If we go all in on always inserting real jobs, we can use the actual client for insert and not have to worry about all the code that's trying to recreate the appropriate default options:

diff --git a/rivertest/worker.go b/rivertest/worker.go
index 947904d..aa7cb12 100644
--- a/rivertest/worker.go
+++ b/rivertest/worker.go
@@ -70,10 +70,29 @@ func NewWorker[T river.JobArgs, TTx any](tb testing.TB, driver riverdriver.Drive
 	}
 }
 
+func (w *Worker[T, TTx]) insertJob(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) *river.Job[T] {
+	tb.Helper()
+
+	result, err := w.client.InsertTx(ctx, tx, args, opts)
+	if err != nil {
+		tb.Fatalf("failed to insert job: %s", err)
+	}
+	var decodedArgs T
+	if err := json.Unmarshal(result.Job.EncodedArgs, &decodedArgs); err != nil {
+		tb.Fatalf("failed to unmarshal args: %s", err)
+	}
+
+	return &river.Job[T]{
+		Args:   decodedArgs,
+		JobRow: result.Job,
+	}
+}
+
 // Work allocates a synthetic job with the provided arguments and optional
 // insert options, then works it.
 func (w *Worker[T, TTx]) Work(ctx context.Context, tb testing.TB, args T, opts *river.InsertOpts) error {
 	tb.Helper()
+
 	job := makeJobFromArgs(tb, w.config, args, opts)
 	return w.WorkJob(ctx, tb, job)
 }
@@ -82,7 +101,8 @@ func (w *Worker[T, TTx]) Work(ctx context.Context, tb testing.TB, args T, opts *
 // insert options, then works it in the given transaction.
 func (w *Worker[T, TTx]) WorkTx(ctx context.Context, tb testing.TB, tx TTx, args T, opts *river.InsertOpts) error {
 	tb.Helper()
-	job := makeJobFromArgs(tb, w.config, args, opts)
+
+	job := w.insertJob(ctx, tb, tx, args, opts)
 	return w.WorkJobTx(ctx, tb, tx, job)
 }

This might not be compatible with the non-tx forms unless we have a way to get the underlying TTx from the ExecutorTx once you GetExecutor().Begin(). But these APIs have existed for a day so I am definitely ok with some fast fixes on this front.

The one thing I can think of that might be hard to test this way is unique jobs: it's conceivable that you could test the same job in two different concurrent tests, and you need to make sure those don't prevent each other from inserting. You'd need to skip uniqueness for these test jobs I think.

[brandur]

Yeah that seems to make sense to me. Enough code smells have dropped out of doing this PR that it seems the original approach could stand to be modified somewhat.

[bgentry]

This is a little unfortunate — you still won't be able to use JobCompleteTx for the Work/WorkTx variants because there's no way to inject the ID of the job which should be updated.

Can you explain a little more what you mean here? Even if it's a synthetic job, JobCompleteTx would still get called—it's just that it will error because the job doesn't actually exist, right? Maybe that was a mistake and we should just go all in w/ inserting real job rows.

[bgentry]

This has been pulled into #766 and it seems like we agree on that as the path forward, so we can close this one. Thanks for driving us toward a better design!