Function Test_Client_InsertTriggersImmediateWork

client_test.go:7361–7423

(t *testing.T)

}

func Test_Client_InsertTriggersImmediateWork(t *testing.T) {
	t.Parallel()

	ctx := context.Background()

	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	t.Cleanup(cancel)

	var (
		dbPool = riversharedtest.DBPool(ctx, t)
		driver = riverpgxv5.New(dbPool)
		schema = riverdbtest.TestSchema(ctx, t, driver, nil)
		config = newTestConfig(t, schema)
	)
	config.FetchCooldown = 20 * time.Millisecond
	config.FetchPollInterval = 20 * time.Second // essentially disable polling
	config.Queues = map[string]QueueConfig{QueueDefault: {MaxWorkers: 1}, "another_queue": {MaxWorkers: 1}}

	type JobArgs struct {
		testutil.JobArgsReflectKind[JobArgs]
	}

	doneCh := make(chan struct{})
	close(doneCh) // don't need to block any jobs from completing
	startedCh := make(chan int64)
	AddWorker(config.Workers, makeAwaitWorker[JobArgs](startedCh, doneCh))

	client := newTestClient(t, dbPool, config)

	startClient(ctx, t, client)
	riversharedtest.WaitOrTimeout(t, client.baseStartStop.Started())

	insertRes, err := client.Insert(ctx, JobArgs{}, nil)
	require.NoError(t, err)

	// Wait for the client to be ready by waiting for a job to be executed:
	select {
	case jobID := <-startedCh:
		require.Equal(t, insertRes.Job.ID, jobID)
	case <-ctx.Done():
		t.Fatal("timed out waiting for warmup job to start")
	}

	// Now that we've run one job, we shouldn't take longer than the cooldown to
	// fetch another after insertion. LISTEN/NOTIFY should ensure we find out
	// about the inserted job much faster than the poll interval.
	//
	// Note: we specifically use a different queue to ensure that the notify
	// limiter is immediately to fire on this queue.
	insertRes2, err := client.Insert(ctx, JobArgs{}, &InsertOpts{Queue: "another_queue"})
	require.NoError(t, err)

	select {
	case jobID := <-startedCh:
		require.Equal(t, insertRes2.Job.ID, jobID)
	// As long as this is meaningfully shorter than the poll interval, we can be
	// sure the re-fetch came from listen/notify.
	case <-time.After(5 * time.Second):