Pattern for aggregating data into a unique job during a scheduled window

[glerchundi]

Hi! First of all, great project. I just discovered it and I really liked the approach.

Please correct me if some of my assumptions are incorrect! Also, is there any other channel for asking these kind of questions? I see no discord, slack, or any other comm. channel in the repo

River unique jobs + scheduled_at pattern is exactly the foundation we need for our event processing system.

Use case

We have a CDC pipeline (outbox -> Pub/Sub -> processors) where multiple events can arrive for the same entity in quick succession. We want to collect all events that arrive within a fixed window and process them together in a single job.

What we'll be doing

We’re using unique jobs with ByPeriod + ScheduledAt to ensure only one job exists per entity during the collection window. That part works perfectly.

The problem is attaching the event data. When the second event arrives and the insert returns UniqueSkippedAsDuplicate, the new event’s args are discarded. We need that data.

Our current workaround is an associated table:

result, err := riverClient.Insert(ctx, ProcessArgs{PartitionKey: "installation:123"}, &river.InsertOpts{
    UniqueOpts: river.UniqueOpts{ByPeriod: 5 * time.Second},
    ScheduledAt: time.Now().Add(5 * time.Second),
})

// Whether new or duplicate, attach the event data
jobID := result.Job.ID
db.Exec("INSERT INTO job_events (job_id, event_id, event_type, data) VALUES ($1, $2, $3, $4)",
    jobID, eventID, eventType, eventData)

When the worker picks up the job, it joins against job_events to get all collected events.

My question:

Is there a recommended pattern within River for this kind of event aggregation? Specifically:

1. Is there a way to append/merge data into an existing unique job’s args rather than discarding the duplicate’s args?
2. If not, is the child table approach reasonable, or is there a better pattern you’d suggest?
3. Any concerns with inserting into a child table referencing result.Job.ID from a UniqueSkippedAsDuplicate result — particularly around race conditions with the worker claiming the job between the insert check and the child row write?

Happy to share more context on our architecture if helpful. Thanks!

(P.D. written in the GitHub mobile app, sorry for the formatting!)

[brandur]

• Is there a way to append/merge data into an existing unique job’s args rather than discarding the duplicate’s args?

Nah afraid not. That insert operation as a DO UPDATE on kind so that a row is always returned, but it won't update args or metadata.

• If not, is the child table approach reasonable, or is there a better pattern you’d suggest?

Yeah, that's the approach I would've recommend after reading the first couple paragraphs of your post. Get different data into a different table and reference that as jobs work.

• Any concerns with inserting into a child table referencing result.Job.ID from a UniqueSkippedAsDuplicate result — particularly around race conditions with the worker claiming the job between the insert check and the child row write?

Nope, I think this should work out reasonably well. Just make sure to do the insert of the job row and your extra row in the same transaction.