Difficulty combining unique jobs and enqueueing next job from within a job

[AntiPaste]

Hi, first of all love the project! I've been using it with both small and massive amounts of jobs and it handles it beautifully.

I'm having some difficulty implementing a specific use case and I hope you have time to guide me. I am using an external API that is cached. It returns an Expires header that describes when the data is refreshed. In order to scrape this efficiently, I want to call this API whenever the data refreshes, but no more than that (barring some rare fault where a job might get re-run).

I have tried multiple approaches but I always end up either having duplicate jobs or missing jobs. The latest attempt I have is using UniqueOpts like so to ensure there is only one non-completed job with the same args (default ByState list except JobStateCompleted removed):

func (ScrapeEntityDataArgs) InsertOpts() river.InsertOpts {
	return river.InsertOpts{
		Queue: QueueScrapeEntityData,
		UniqueOpts: river.UniqueOpts{
			ByArgs: true,
			ByState: []rivertype.JobState{
				rivertype.JobStateAvailable,
				rivertype.JobStatePending,
				rivertype.JobStateRunning,
				rivertype.JobStateRetryable,
				rivertype.JobStateScheduled,
			},
		},
	}
}

There is a periodic job that inserts all the necessary ScrapeEntityData jobs to kickstart this. The job re-inserts itself, scheduled for the next data refresh like so:

scheduledAt := expires.Add(1 * time.Second)
if _, err := river.JobCompleteTx[*riverpgxv5.Driver](ctx, tx, job); err != nil {
	return fmt.Errorf("failed to complete job: %w", err)
}

if err := tx.Commit(ctx); err != nil {
	return fmt.Errorf("failed to commit transaction: %w", err)
}

client := river.ClientFromContext[pgx.Tx](ctx)
if _, err := client.Insert(ctx, ScrapeEntityDataArgs{EntityID: job.Args.EntityID}, &river.InsertOpts{ScheduledAt: scheduledAt}); err != nil {
	w.logger.Error().Err(err).Msg("failed to insert job")
}

For some reason the job insertion is skipped as duplicate. I think there is some kind of finalization aspect where even though I'm calling JobCompleteTx it's not removed from consideration when checking duplicates until Work returns.

Other approaches I've tried:

• no UniqueOpts on job level, but UniqueOpts on periodic insert. This will cause hard-to-detect duplicate jobs if the job ever gets run twice or more

[AntiPaste]

Looks like this ended up being user error. I had only ByArgs when inserting the periodic job so the unique check would fail as expected. Thanks again for the project.

[brandur]

No problem Kasper. Glad you got it figured out!