brandur
commented
3 months ago Hm, okay, quite a lot of content here, but speaking to the headline issue: I spoke to @bgentry a little about this, and it does feel like a per-job customizable RescueStuckJobsAfter is just a little too specific of a feature for anything even close to a general case. As in, I'd expect 99.99% of users wouldn't need it.
One point of clarification: ideally in a perfectly functioning system, RescueStuckJobsAfter is never used. Remember that this is basically only there to cover the case of an unexpected crash that might occur during something like a panic. If jobs return an error instead like they're supposed to, they're rescheduled immediately, and never need to be rescued.
a
deadlinefeature inInsertOptsi think would make me really happy, as it would also resolve other issues im having to work around. workers should discard jobs that they pick up that are past their deadline, and a maintenance job which "discards" tasks that are after their deadline could run on a regular interval.
This is interesting, but my first reaction against is somewhat similar to the above: it'd be such a specific feature that I think we'd want some very strong rationale for why it should be included in the default API.
It's worth remembering too that River could be quite easily be augmented with something like a common job bootstrap where project-specific functionality could be customized to your heart's desire. e.g.
type MyProjectCommonArgs struct {
Deadline time.Time `json:"deadline"`
}
func (a *MyProjectCommonArgs) GetCommonArgs() *MyProjectCommonArgs {
return a
}
type WithCommonArgs interface {
GetCommonArgs() *MyProjectCommonArgs
}
type SortArgs struct {
MyProjectCommonArgs
// Strings is a slice of strings to sort.
Strings []string `json:"strings"`
}
func (SortArgs) Kind() string { return "sort" }func CommonJobBootstrap[T WithCommonArgs](ctx context.Context, args T, workFunc func(args T) error) error {
commonArgs := args.GetCommonArgs()
if commonArgs.Deadline.Before(time.Now()) {
return errors.New("deadline exceeded")
}
return workFunc(args)
}
type SortWorker struct {
// An embedded WorkerDefaults sets up default methods to fulfill the rest of
// the Worker interface:
river.WorkerDefaults[SortArgs]
}
func (w *SortWorker) Work(ctx context.Context, job *river.Job[SortArgs]) error {
return CommonJobBootstrap(ctx, args, func(args SortArgs) error {
sort.Strings(job.Args.Strings)
fmt.Printf("Sorted strings: %+v\n", job.Args.Strings)
return nil
}
}
elee1766
bgentry
we have some jobs that take 1-2 minutes, and other jobs that takes at most 5 seconds, and so we set worker timeouts accordingly
however, we can't configure RescueStuckJobsAfter for an individual job. this means that when a worker running the 5 seconds task dies, we need to wait for the global RescueStuckJobsAfter, instead of what should be 5-10 seconds. this means that updates that should be happening around once a second pause for a few minutes instead of a few seconds.
we schedule with a unique constraint because this task runs on every event, but sometimes events can be up to 10 per second, and we are okay with not scheduling the task if it is already running since it is an aggregate and will just be picked up on the next unique task. if we schedule too many jobs, we will not only schedule more than we can process, but be doing extra work, and need to worry about race conditions where tasks that started earlier finish later.
one possible solution is to read the
jobstable during InsertTx, and check usingscheduled_timeif the job is stuck ourselves, and manually fix it, but i'm not sure if that's allowed. for now i am checking the scheduled_time in a loop in my go application to manually rescue the task as this one specific task is critical to the movement of the rest of the applications.a
deadlinefeature inInsertOptsi think would make me really happy, as it would also resolve other issues im having to work around. workers should discard jobs that they pick up that are past their deadline, and a maintenance job which "discards" tasks that are after their deadline could run on a regular interval.the problem is adjacent to something that @bgentry mentioned in a previous reply to me in https://github.com/riverqueue/river/issues/336#issuecomment-2093895895 . This is actually a different summary job in a different app, we deal with a lot of data that needs to be updated in this manner.
we have had success using some hacky locking mechanisms for our other tasks, however in this case we are trying using the job=running as our locking mechanism, and the issue is that it takes too long for a dying worker to "abandon" their lock.
for reference, the system we are migrating from used a user defined hash as an optional uniqueness key with support for a job timeout and deadline on the job level. the key would be locked when scheduled, and unlocked at deadline or job completion. we could bring that idea back and use it along with river.
this calls back to what i was told to do here: https://github.com/riverqueue/river/discussions/346
and also very much relates to this problem: https://github.com/riverqueue/river/issues/165
just checking if the job is running using my own locks table.
that said, these restrictions + the incredibly slow insert speed (100-200rps) with unique makes me feel like the current "unique constraint" feature is missing a lot of power.
I am feeling like my best course of action at this point given the current state of Unique to move forward with river is to build a layer on top of river that deals a little better with these more complex scheduling constraints.