Updating job metadata while running

[atoi]

I use jobs to perform unique short-to-medium-running tasks. The tasks are not just some background stuff, but something a user can observe.
The nature of a task lets me roughly estimate its progress (I paginate through an API), so I'm considering writing that progress to job metadata column once in a while. The reason is showing a rough progress estimation is much better user experience than just displaying a spinner for indefinite amount of time.

Yes, I am aware river API does not let me do that, but I have no problem with doing that myself. Just wanted to ask if that's a good idea and what are the chances of me breaking river job processing logic if I do so.

[bgentry]

Hi @atoi, good question. If you’re going to do this yourself anyway (presumably as a query invoked within the job) the safer thing to do is probably to update the job’s args instead. That also feels more appropriate in this case because it sounds like this is a job-specific value that you’re tracking, whereas metadata is going to be used more for layered on functionality that isn’t job-specific.

You can use jsonb operators to safely access/set subfields within your json structure. Up to you of course not to accidentally clobber your existing args 😆 As long as your args struct includes something like a progress field you should be able to safely update it at the end of the job before returning.

A more formal way of doing this would be to have your own table for these operations where you record & coordinate their progress, although sometimes this lighter weight option is ok too. An example of what this could look like: say you were zipping up a bunch of files for the user to download. If you had another table like file_downloads to serve as the source of truth for this operation, you could easily update progress there. Or you could even use listen/notify or some other pubsub system to broadcast progress that the client can listen to.

There’s no real risk in affecting job processing with any of these options, though, as long as you don’t hold open a super long transaction on the row or forget to close one. Best to make sure you have the appropriate default transaction timeout set on your pgxpool just in case your worker dies midway through the update.

[atoi]

Thank you for the thorough explanation!

Yes, I was debating on whether to use args or metadata and opted for metadata. Well, as you say metadata is kind of reserved for features to come, I suppose I have to stick to args.
See, the reasoning was args are "things a job needs to get started" and metadata is "a bag of stuff we don't know where to put". But not only that -- I want the jobs to be unique and putting progress into args will break ByArgs: true criteria.

I suppose, the safest way would be to add a separate table to track progress. I hoped I could avoid that by piggy-backing on already existing river functionality 😆

[atoi]

In case I decide to stay lazy and abuse the current river_job table, I suppose I should be fine if I just modify a single field and not replace the whole meta? To be even safer, I could add a special namespace that is unlikely to be used by river.

-- name: SetJobMetaProgress :exec
UPDATE job_queues."river_job"
SET "metadata" = jsonb_set("metadata", '{progress}', to_jsonb(sqlc.arg(progress)::INT), true)
WHERE "id" = sqlc.arg(job_id);

[bgentry]

That would work, although if you’re going to use that field I would suggest a slightly less generic name to avoid potential conflicts. A namespace or prefix as you said should be ok.

And good point about my suggestion interfering with arg uniqueness. If you keep it on the jobs table here there’s also the additional bloat and read/write throughout to keep in mind, but that’ll only be an issue if you’re running at higher scale with many jobs per second.

[atoi]

Thank you again. This conversation is really helpful!

I suppose, I'll just add a collision-safe-ish namespace to metadata then.
As for the read/write throughout -- I don't expect that becoming a problem anytime soon.

[vsoch]

This discussion was really helpful! For another idea (because I struggled with this for a few hours) what I got working from the worker is this query:

UpdateNodesQuery  = "UPDATE river_job SET args = jsonb_set(args, '{nodes}', to_jsonb($1::text)) WHERE id=$2;"

I tried other derivations of that jsonb_set and always got some error, and I needed the ::text and provided like that because it was fussy about the type. Then in the worker I wound up creating the pool again, and doing:

// We must update the database with nodes from here with a query
// This will be sent back to the Kubernetes scheduler
pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
	return err
}

_, err = pool.Query(ctx, queries.UpdateNodesQuery, nodeStr, job.ID)
if err != nil {
	return err
}

For some context - my goal is to send the node list (a worker result) back to one of the event subscribers (the main schedule loop) so I'm going to look at parsing the output there next. The strategy above is adding another Arg "nodes" that starts empty and is updated after a successful allocation.

I can see it in the print and it looks like it is json serialized a few times (lots of escaping going on!) I'll update the post here with what I wind up doing, because I don't think there is a way to persist results.