Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion internal/maintenance/job_rescuer.go
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,7 @@ func (s *JobRescuer) runOnce(ctx context.Context) (*rescuerRunOnceResult, error)
}

func (s *JobRescuer) getStuckJobs(ctx context.Context) ([]*rivertype.JobRow, error) {
ctx, cancelFunc := context.WithTimeout(ctx, 30*time.Second)
ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
defer cancelFunc()

stuckHorizon := time.Now().Add(-s.Config.RescueAfter)
Expand Down
2 changes: 1 addition & 1 deletion internal/maintenance/job_scheduler.go
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ func (s *JobScheduler) runOnce(ctx context.Context) (*schedulerRunOnceResult, er
for {
// Wrapped in a function so that defers run as expected.
numScheduled, err := func() (int, error) {
ctx, cancelFunc := context.WithTimeout(ctx, 30*time.Second)
ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
defer cancelFunc()

tx, err := s.exec.Begin(ctx)
Expand Down
2 changes: 1 addition & 1 deletion internal/maintenance/queue_cleaner.go
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,7 @@ func (s *QueueCleaner) runOnce(ctx context.Context) (*queueCleanerRunOnceResult,
for {
// Wrapped in a function so that defers run as expected.
queuesDeleted, err := func() ([]string, error) {
ctx, cancelFunc := context.WithTimeout(ctx, 30*time.Second)
ctx, cancelFunc := context.WithTimeout(ctx, riversharedmaintenance.TimeoutDefault)
defer cancelFunc()

queuesDeleted, err := s.exec.QueueDeleteExpired(ctx, &riverdriver.QueueDeleteExpiredParams{
Expand Down
3 changes: 2 additions & 1 deletion rivershared/circuitbreaker/circuit_breaker.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,8 @@ func (b *CircuitBreaker) ResetIfNotOpen() bool {

// Trip "trips" the circuit breaker by counting an action towards opening it. If
// the action causes the breaker to reach its limit within its window, the
// breaker opens and the function returns true.
// breaker opens and the function returns true. Subsequent calls to Trip after
// the breaker is open will also return true.
func (b *CircuitBreaker) Trip() bool {
if b.open {
return true
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,12 @@ const (
LogPrefixRanSuccessfully = ": Ran successfully"
LogPrefixRunLoopStarted = ": Run loop started"
LogPrefixRunLoopStopped = ": Run loop stopped"

// TimeoutDefault is a reasonable timeout for any large maintenance-related
// queries. Some maintainers may opt to switch to their own timeout, but
// this one should generally be used unless there's a good reason to have a
// specific version.
TimeoutDefault = 30 * time.Second
)

// Constants related to JobCleaner.
Expand Down
Loading