Add timeout to standard pilot fetch

[peter941221]

Summary

Fix StandardPilot.JobGetAvailable so a stalled fetch does not hang a producer indefinitely.

Problem

producer.dispatchWork intentionally strips cancellation from the work context before fetching jobs so an in-flight fetch is allowed to complete during shutdown:

• producer.go:744-766

That is reasonable, but StandardPilot.JobGetAvailable forwarded directly to exec.JobGetAvailable with no timeout at all:

• rivershared/riverpilot/standard_pilot.go:18-22

This meant a stalled driver call could block a standard-pilot producer forever. The pro pilot already applies per-attempt fetch timeouts, so the standard pilot was the outlier.

Change

Add a 10-second timeout inside StandardPilot.JobGetAvailable before calling the driver.

This keeps the existing shutdown semantics intact:

• fetches still ignore parent cancellation from dispatchWork
• but they are now bounded, so a wedged DB call eventually returns instead of freezing the producer forever

The timeout is local to the standard pilot so there is no driver SQL change and no producer state-machine change.

Testing

• added rivershared/riverpilot/standard_pilot_test.go
• covered MaxToLock <= 0 no-op behavior
• covered a hung JobGetAvailable call timing out with context.DeadlineExceeded
• covered parent cancellation still winning when the incoming context is already canceled

Verification

Locally verified with:

• GOPROXY=https://goproxy.cn,direct GOSUMDB=off go test ./rivershared/riverpilot -count=1

Closes #1026.

[brandur]

Thanks!

@bgentry Any strong opinions on how you want to handle this one? Another option is to just put the timeout in producer.go in dispatchWork:

func (p *producer) dispatchWork(workCtx context.Context, count int, fetchResultCh chan<- producerFetchResult) {
	// This intentionally removes any deadlines or cancellation from the parent
	// context because we don't want it to get cancelled if the producer is asked
	// to shut down. In that situation, we want to finish fetching any jobs we are
	// in the midst of fetching, work them, and then stop. Otherwise we'd have a
	// risk of shutting down when we had already fetched jobs in the database,
	// leaving those jobs stranded. We'd then potentially have to release them
	// back to the queue.
	ctx := context.WithoutCancel(workCtx)

	// Maximum size of the `attempted_by` array on each job row. This maximum is
	// rarely hit, but exists to protect against degenerate cases.
	const maxAttemptedBy = 100

	jobs, err := p.pilot.JobGetAvailable(ctx, p.exec, p.state, &riverdriver.JobGetAvailableParams{
		ClientID:       p.config.ClientID,
		MaxAttemptedBy: maxAttemptedBy,
		MaxToLock:      count,
		Now:            p.Time.NowOrNil(),
		Queue:          p.config.Queue,
		ProducerID:     p.id.Load(),
		Schema:         p.config.Schema,
	})
	if err != nil {
		fetchResultCh <- producerFetchResult{err: err}
		return
	}

	fetchResultCh <- producerFetchResult{jobs: jobs}
}

That might be better in the way that not every pilot needs to remember to bring its own context cancellations. That said, maybe in this case we might want a longer cancellation for the pro pilot so it'd make sense to break up the two.