Worker is trying to rescue jobs from queues that are not assigned to it

A bit of context

We have two kinds of workers at the moment, sharing the same database, let's say worker-1 is running jobs from queue-1 and worker-2 is running jobs from queue-2. However, we are using the same binary for the worker and we recently tried registering jobs conditionally, basically registering only those jobs that are supposed to be executed by this worker based on queue selection.

Another thing to mention is that our RescueStuckJobsAfter is configured to be pretty small (30s) whereas timeouts on the jobs are minutes.

An issue

After introducing this conditional registration, we observed the following behaviour:

worker-1 is fetching all jobs (for both queues) to rescue:

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

Lines 257 to 263 in eb0b985

    
           -- name: JobGetStuck :many 
        
           SELECT * 
        
           FROM /* TEMPLATE: schema */river_job 
        
           WHERE state = 'running' 
        
               AND attempted_at < @stuck_horizon::timestamptz 
        
           ORDER BY id 
        
           LIMIT @max;

amongst others, it can fetch jobs dedicated for queue-2 and they are not registered in the worker-1, so makeRetryDecision discards the job:

river/internal/maintenance/job_rescuer.go

Lines 302 to 307 in eb0b985

    
           workUnitFactory := s.Config.WorkUnitFactoryFunc(job.Kind) 
        
           if workUnitFactory == nil { 
        
           	s.Logger.ErrorContext(ctx, s.Name+": Attempted to rescue unhandled job kind, discarding", 
        
           		slog.String("job_kind", job.Kind), slog.Int64("job_id", job.ID)) 
        
           	return jobRetryDecisionDiscard, time.Time{} 
        
           }

Checked documentation if we are missing something, and both pages that could have mention any details of this behaviour are not saying anything about our weird case:

Inserting and working jobs does not mention if all jobs from a single database have to be registered in all workers
Multiple queues only says but workers will only select jobs to work for queues that they're configured to handle, but still it is unclear if we should register all jobs or not

A very bad side effect of this for us was that discarded job continued execution (context wasn't cancelled) and another job with the same unique_key was scheduled, violating unique constraint that we are relying on for correctness, but I am not blaming this on river, this is a consequence of our ignorance of rescue mechanics nuances.

Short-term we fixed our issue by again registering all jobs in the worker binary regardless its configuration, but it would be valuable to have an answer if this is a designed behaviour or a bug?

@bgentry

Yeah, for better or worse, this is currently implemented as originally designed.

I've thought about modifying the various maintenance services so they only work on jobs that their associated client knows about, thereby making isolation possible between multiple clients trying to make separate use of the same schema. I think this isn't a bad idea, but it does have other downsides. For example, jobs and/or queues which have become "orphaned" (i.e. they were in use previously, but no client is configured with them anymore) will not get any attention from any clients. It's also a bit of a non-trivial change because we'd have to account for multiple leadership mechanisms that can be configured independently.

For now, a quick fix for what you're trying to do is to switch to using multiple schemas.

I'll flag this one for discussion with @bgentry in case there's anything else we could potentially be doing here.

"orphaned" will not get any attention from any clients

Yeah, that makes sense, I was also thinking about it.

Then I think action items on this one could be:

Mention this in both documentation pages on registering jobs and multiple queues
Cancel context by the job itself if the job was marked as discarded in the db (I am not sure though if there is any active polling from the side of running job now)

Add leadership "domains" so multiple Rivers can operate in one schema

@NanoBjorn

@NanoBjorn I ended up putting together a change that enables multiple leadership "domains" inside of a single schema in #1113. Do you want to take a look at that and see if it'd lend itself to what you were trying to do here?

Add leadership "domains" so multiple Rivers can operate in one schema

It looks like LeaderDomain is introducing additional complexity in several places and increasing the cognitive load for anyone configuring it. I’m not sure whether it’s worth solving the problem we’re trying to address this way, the cognitive cost looks a little bit too high; perhaps simply documenting these nuances would be sufficient, along with fixing context cancellation for discarded jobs.

@brandur

Hi @brandur and @bgentry we experienced an issue today directly related to this. A few months ago we migrated from running 9 different river schemas for each of our applications to running a single river schema because the overhead of managing an increasing number of river schemas was adding operational overhead and baggage to our team's workflows. We did this because of the Durable Periodic Job feature that was available on RiverPro which allowed us to get around issues as described in this previous discussion. We overlooked the fact that the elected leader when it tries to "rescue" stuck jobs would see many jobs not part of its own Worker bundle and immediately Discard the job without any warning or firing any event other than emitting an Error log "Attempted to rescue unhandled job kind, discarding".

I understand why this behavior is the way it is but we are now faced with needing to solve for this problem and going back to the n-schema approach is really not-ideal. The code in job_rescuer.go in makeRetryDecision seems to just require the worker bundle to be able to make a decision on whether to ignore and if/when to retry the job. I wonder if there's a tactical solution that would allow the rescuer to take a more lenient approach and retry the job.

I haven't looked much into the above Leadership Domains PR but it seems like the work there stalled out because the complexity of the code and implications with other River behaviors. Do you have any idea for what the right long term solution is here?

tagging @jackHedaya

I wonder if there's a way to configure the Job definitions universally across all River clients but not all the Jobs are part of the client's Worker bundle. Obviously would be a bit of a change to core River, but curious if there's a way to differentiate between "a list of all the jobs managed within the schema" and a "list of jobs worked by a given client" (e.g. Worker bundle). Curious to hear your thoughts on whether this is feasible.

@bgentry

Hm, it seems like the check that disallows unknown jobs is that one that you should be able to disable by configuration option, and maybe even one that we could remove completely. Let me discuss with @bgentry a bit.

@magaldima

@magaldima Hey, I was just talking this through with Blake and I think we have some ideas, but we wanted to make sure that we're properly understanding your use case before going forward with anything.

Would it be accurate to say:

You have multiple independent applications sharing a single database/schema.
The different applications use different River queues (i.e. they're not all trying to share the default queue).

@brandur

@brandur I can confirm both. We have a single shared schema across applications, and each application owns and works only its own queues.

Add leadership "domains" so multiple Rivers can operate in one schema

@bgentry

Ended up resurrecting #1113 for some more discussion. Talked this through yesterday with @bgentry and unfortunately we couldn't come up with too many good alternatives if we want to support this use case.

Any progress on solving this?

We are trying to migrate from the old que_jobs library to riverqueue. Similar setup with multiple application modules running separately, each has their own schema for its own data and a shared messaging schema to send jobs between modules.

Multiple riverqueue schemas wouldn't work as the messages have to go between modules.

Effectively this is blocking the migration.

On another note: schema name validation is currently weird as well. Not allowing perfectly valid schema names in a postgres db containing "-" character. Schema names in queries should be properly quoted when worried about characters that may have multiple meanings (

river/client.go

Line 573 in 78ae535

if c.Schema != "" && !postgresSchemaNameRE.MatchString(c.Schema) {

).

@magaldima

Following up on @magaldima's post above. We ended up implementing the following solution to prevent the river leader from discarding jobs "owned" by other services in our setup.

To reiterate, our system maintains the following invariants in our river setup.

Each service has a set of queues which it owns and no other service works.
Each job kind is associated with a single queue.
All services share a single river schema.

We kept the single shared schema and instead register all job kinds with all clients across our services. Each client registers a "real" worker for the job kinds it owns, and a stub worker for the job kinds it does not own.

type stubWorker[T river.JobArgs] struct {
	river.WorkerDefaults[T]
}

func (w *stubWorker[T]) Work(ctx context.Context, job *river.Job[T]) error {
	// Unreachable in normal operation: this client doesn't subscribe to the
	// queue this job is on.
	reportUnownedJob(job.JobRow)

	// Snoozing doesn't consume an attempt, so the job survives until the
	// setup issue is fixed and the client that owns it picks it up.
	return river.JobSnooze(time.Hour)
}

Because every job kind is registered with every client, the rescuer behavior for any given job kind is no longer dependent on which client happens to be the leader.

There are some awkward parts of this setup. We rely on a lint rule requiring every JobArgs type to be declared in one shared package, and a test that fails when a kind in that package has no stub registered. If an improper setup makes it into the code, we rely on error reporting and the job snooze to help us get the system back into a good state. However, it's worked well so far and has eliminated the more serious issue of jobs being discarded silently.

cc @jackHedaya

Navigation Menu

Repository navigation

Worker is trying to rescue jobs from queues that are not assigned to it #1105

A bit of context

An issue

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	-- name: JobGetStuck :many
	SELECT *
	FROM /* TEMPLATE: schema */river_job
	WHERE state = 'running'
	AND attempted_at < @stuck_horizon::timestamptz
	ORDER BY id
	LIMIT @max;

	workUnitFactory := s.Config.WorkUnitFactoryFunc(job.Kind)
	if workUnitFactory == nil {
	s.Logger.ErrorContext(ctx, s.Name+": Attempted to rescue unhandled job kind, discarding",
	slog.String("job_kind", job.Kind), slog.Int64("job_id", job.ID))
	return jobRetryDecisionDiscard, time.Time{}
	}

Worker is trying to rescue jobs from queues that are not assigned to it #1105

Description

A bit of context

An issue

Activity

brandur commented on Dec 18, 2025

NanoBjorn commented on Dec 18, 2025

brandur commented on Dec 24, 2025

NanoBjorn commented on Jan 9, 2026

magaldima commented on Apr 28, 2026

magaldima commented on Apr 28, 2026

brandur commented on Apr 28, 2026

brandur commented on Apr 30, 2026

jackHedaya commented on Apr 30, 2026

brandur commented on May 1, 2026

toomy4 commented on Jul 1, 2026

themaxgoldman commented on Sep 8, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions