Skip to content

Commit da2bc55

Browse files
committed
Add Config.StopAbandonTimeout to abandon jobs still running after stop timeouts
Here, add a new `Config.StopAbandonTimeout` on top of the existing `SoftStopTimeout` whose job it is to recover badly behaving job as much as possible before coming to a full stop. Currently, if a client is stopping and is running jobs that don't respond to context cancellation, those jobs end up getting left in a `running` state, which means that they won't be recoverable again until they're rescued an hour later. `StopAbandonTimeout` engages after soft stop, and has each producer initiate abandonment of its running jobs, which means setting them to an error state (or discarded in case a job is out of retries). Because they're errored, they'll get to run immediately the next time a client starts up. Ideally, users don't need to depend on this functionality since the "correct" behavior would be to make sure that all jobs are able to respond to context cancellation, so we make this new feature optional.
1 parent fdf6cdf commit da2bc55

8 files changed

Lines changed: 773 additions & 66 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
99

1010
### Added
1111

12+
- Added `Config.StopAbandonTimeout` to bound how long a client waits after job contexts are cancelled during shutdown. When the timeout elapses, jobs that have not begun finalizing are abandoned, recorded as failed, and set as errored (or discarded if retries are exhausted). Worker goroutines cannot be forcibly terminated and continue running until the process exits. [PR #1289](https://github.com/riverqueue/river/pull/1289).
1213
- Added `EventKindJobInterrupted`, emitted when a running job is interrupted because its client is shutting down, the job was cancelled, and has been made immediately available to be worked again. [PR #1290](https://github.com/riverqueue/river/pull/1290).
1314

1415
### Changed

‎client.go‎

Lines changed: 103 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -378,11 +378,9 @@ type Config struct {
378378
Schema string
379379

380380
// SoftStopTimeout is the maximum amount of time that the client will wait
381-
// for running jobs to finish during a stop before their contexts are
382-
// cancelled. After the timeout elapses, the client escalates to a hard stop
383-
// by cancelling the context of all running jobs. This applies regardless of
384-
// how stop is initiated — whether by calling Stop, StopAndCancel, or by
385-
// cancelling the context passed to Start.
381+
// for running jobs to finish during a graceful stop before entering soft
382+
// stop by cancelling job contexts. This applies when stop is initiated by
383+
// calling Stop or by cancelling the context passed to Start.
386384
//
387385
// In combination with signal.NotifyContext on the context passed to Start,
388386
// this can simplify graceful stop to:
@@ -393,16 +391,33 @@ type Config struct {
393391
// if err := client.Start(ctx); err != nil { ... }
394392
// <-client.Stopped()
395393
//
396-
// The signal cancels the Start context, which initiates a soft stop. If
394+
// The signal cancels the Start context, which initiates a graceful stop. If
397395
// running jobs haven't finished after SoftStopTimeout, their contexts are
398-
// automatically cancelled to trigger a hard stop.
396+
// cancelled.
399397
//
400-
// StopAndCancel bypasses the timeout entirely and cancels job contexts
401-
// immediately.
398+
// StopAndCancel cancels job contexts immediately instead of waiting for
399+
// SoftStopTimeout.
402400
//
403401
// Defaults to no timeout (wait indefinitely for jobs to finish).
404402
SoftStopTimeout time.Duration
405403

404+
// StopAbandonTimeout is the maximum amount of time that the client will wait
405+
// after job contexts are cancelled during a shutdown "soft stop" before it
406+
// abandons jobs still running (i.e. those which did not respond to context
407+
// cancellation). Abandoned jobs are recorded as failed and made immediately
408+
// available for retry, or discarded if they have no attempts remaining.
409+
//
410+
// Worker goroutines cannot be forcibly terminated, and may continue running
411+
// after their jobs are abandoned. Job implementations should therefore be
412+
// prepared for duplicate execution.
413+
//
414+
// The timer starts only after a soft stop has begun by cancelling job
415+
// contexts, like after SoftStopTimeout elapses, StopAndCancel is called, or
416+
// the Start context is cancelled without SoftStopTimeout configured.
417+
//
418+
// Defaults to no timeout (job abandonment disabled).
419+
StopAbandonTimeout time.Duration
420+
406421
// SkipJobKindValidation causes the job kind format validation check to be
407422
// skipped. This is available as an interim stopgap for users that have
408423
// invalid job kind names, but would rather disable the check rather than
@@ -536,6 +551,7 @@ func (c *Config) WithDefaults() *Config {
536551
RetryPolicy: retryPolicy,
537552
Schema: c.Schema,
538553
SoftStopTimeout: c.SoftStopTimeout,
554+
StopAbandonTimeout: c.StopAbandonTimeout,
539555
SkipJobKindValidation: c.SkipJobKindValidation,
540556
SkipUnknownJobCheck: c.SkipUnknownJobCheck,
541557
Test: c.Test,
@@ -566,6 +582,9 @@ func (c *Config) validate() error {
566582
if c.FetchPollInterval < c.FetchCooldown {
567583
return fmt.Errorf("FetchPollInterval cannot be shorter than FetchCooldown (%s)", c.FetchCooldown)
568584
}
585+
if c.StopAbandonTimeout < 0 {
586+
return errors.New("StopAbandonTimeout cannot be less than zero")
587+
}
569588
if len(c.ID) > 100 {
570589
return errors.New("ID cannot be longer than 100 characters")
571590
}
@@ -601,6 +620,9 @@ func (c *Config) validate() error {
601620
if c.Schema != "" && !postgresSchemaNameRE.MatchString(c.Schema) {
602621
return errors.New("Schema name can only contain letters, numbers, and underscores, and must start with a letter or underscore")
603622
}
623+
if c.SoftStopTimeout < 0 {
624+
return errors.New("SoftStopTimeout cannot be less than zero")
625+
}
604626

605627
for queue, queueConfig := range c.Queues {
606628
if err := queueConfig.validate(queue, c.FetchCooldown, c.FetchPollInterval); err != nil {
@@ -1069,10 +1091,12 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
10691091
// A graceful shutdown stops fetching new jobs but allows any previously fetched
10701092
// jobs to complete. This can be initiated with the Stop method.
10711093
//
1072-
// A more abrupt shutdown can be achieved by either cancelling the provided
1073-
// context or by calling StopAndCancel. This will not only stop fetching new
1074-
// jobs, but will also cancel the context for any currently-running jobs. If
1075-
// using StopAndCancel, there's no need to also call Stop.
1094+
// A soft stop cancels job contexts after fetching has stopped. It can be
1095+
// initiated by calling StopAndCancel, by cancelling the provided context when
1096+
// SoftStopTimeout is not configured, or by waiting for SoftStopTimeout to elapse
1097+
// during graceful stop. If StopAbandonTimeout is configured, jobs still running
1098+
// after that timeout will be abandoned and recorded as failed. If using
1099+
// StopAndCancel, there's no need to also call Stop.
10761100
func (c *Client[TTx]) Start(ctx context.Context) error {
10771101
fetchCtx, shouldStart, started, stopped := c.baseStartStop.StartInit(ctx)
10781102
if !shouldStart {
@@ -1086,9 +1110,13 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
10861110
// sure to take a channel reference before finishing stopped.
10871111
c.stopped = c.baseStartStop.StoppedUnsafe()
10881112

1089-
producersAsServices := func() []startstop.Service {
1113+
producers := func() []*producer {
1114+
return maputil.Values(c.producersByQueueName)
1115+
}
1116+
1117+
producersAsServices := func(producers []*producer) []startstop.Service {
10901118
return sliceutil.Map(
1091-
maputil.Values(c.producersByQueueName),
1119+
producers,
10921120
func(p *producer) startstop.Service { return p },
10931121
)
10941122
}
@@ -1142,8 +1170,8 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
11421170
// We use separate contexts for fetching and working to allow for a
11431171
// graceful stop. When SoftStopTimeout is configured, the work context
11441172
// is detached from the start context so that cancelling the start
1145-
// context initiates a soft stop (with timeout escalation) rather than
1146-
// an immediate hard stop. When SoftStopTimeout is not configured, the
1173+
// context initiates a graceful stop (with timeout escalation) rather
1174+
// than an immediate soft stop. When SoftStopTimeout is not configured, the
11471175
// work context inherits from the start context to preserve the
11481176
// existing behavior where cancelling the start context is equivalent
11491177
// to StopAndCancel.
@@ -1166,7 +1194,7 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
11661194
for _, producer := range c.producersByQueueName {
11671195
if err := producer.StartWorkContext(fetchCtx, workCtx); err != nil {
11681196
workCancel(err)
1169-
startstop.StopAllParallel(producersAsServices()...)
1197+
startstop.StopAllParallel(producersAsServices(producers())...)
11701198
stopServicesOnError()
11711199
return err
11721200
}
@@ -1188,7 +1216,7 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
11881216
// Generate producer services while c.queues.startStopMu.Lock() is still
11891217
// held. This is used for WaitAllStarted below, but don't use it elsewhere
11901218
// because new producers may have been added while the client is running.
1191-
producerServices := producersAsServices()
1219+
producerServices := producersAsServices(producers())
11921220

11931221
go func() {
11941222
// Wait for all subservices to start up before signaling our own start.
@@ -1215,22 +1243,57 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
12151243
c.queues.startStopMu.Lock()
12161244
defer c.queues.startStopMu.Unlock()
12171245

1246+
producerList := producers()
1247+
1248+
abandonTimerCtx, abandonTimerCancel := context.WithCancel(context.WithoutCancel(ctx))
1249+
defer abandonTimerCancel()
1250+
1251+
startAbandonTimer := sync.OnceFunc(func() {
1252+
if c.config.StopAbandonTimeout <= 0 {
1253+
return
1254+
}
1255+
1256+
go func() {
1257+
timer := time.NewTimer(c.config.StopAbandonTimeout)
1258+
defer timer.Stop()
1259+
1260+
select {
1261+
case <-timer.C:
1262+
c.baseService.Logger.WarnContext(ctx, c.baseService.Name+": StopAbandonTimeout elapsed; abandoning remaining jobs", slog.Duration("stop_abandon_timeout", c.config.StopAbandonTimeout))
1263+
for _, producer := range producerList {
1264+
producer.abandon()
1265+
}
1266+
case <-abandonTimerCtx.Done():
1267+
}
1268+
}()
1269+
})
1270+
1271+
workCtx := c.queues.workCtx
1272+
go func() {
1273+
select {
1274+
case <-workCtx.Done():
1275+
startAbandonTimer()
1276+
case <-abandonTimerCtx.Done():
1277+
}
1278+
}()
1279+
12181280
// If SoftStopTimeout is configured, start a timer that will cancel
1219-
// the work context (escalating to a hard stop) if producers don't
1220-
// finish in time. StopAndCancel also calls workCancel, in which case
1221-
// this timer is a harmless no-op because the context is already done.
1281+
// the work context if producers don't finish in time. Once the work
1282+
// context is cancelled, the optional abandonment timer starts.
12221283
if c.config.SoftStopTimeout > 0 {
12231284
softStopTimer := time.AfterFunc(c.config.SoftStopTimeout, func() {
12241285
c.baseService.Logger.WarnContext(ctx, c.baseService.Name+": Soft stop timeout; cancelling remaining job contexts", slog.Duration("soft_stop_timeout", c.config.SoftStopTimeout))
12251286
c.workCancel(rivercommon.ErrStop)
1287+
startAbandonTimer()
12261288
})
12271289
defer softStopTimer.Stop()
12281290
}
12291291

12301292
// On stop, have the producers stop fetching first of all.
12311293
c.baseService.Logger.DebugContext(ctx, c.baseService.Name+": Stopping producers")
1232-
startstop.StopAllParallel(producersAsServices()...)
1294+
startstop.StopAllParallel(producersAsServices(producerList)...)
12331295
c.baseService.Logger.DebugContext(ctx, c.baseService.Name+": All producers stopped")
1296+
abandonTimerCancel()
12341297

12351298
c.workCancel(rivercommon.ErrStop)
12361299

@@ -1259,12 +1322,17 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
12591322
// complete before exiting. If the provided context is done before shutdown has
12601323
// completed, Stop will return immediately with the context's error.
12611324
//
1262-
// If SoftStopTimeout is configured, running job contexts will be automatically
1263-
// cancelled after the timeout elapses, escalating to a hard stop. This also
1264-
// applies when stop is initiated by cancelling the context passed to Start.
1325+
// If SoftStopTimeout is configured, jobs still running after the timeout
1326+
// elapses have their contexts cancelled.
1327+
//
1328+
// If StopAbandonTimeout is configured, jobs still running after SoftStopTimeout
1329+
// and StopAbandonTimeout have elapsed (i.e. waited for jobs to stop gracefully
1330+
// before cancelling, then waited again for them to stop on cancel) are abandoned
1331+
// and recorded as failed so they can be retried immediately. This also applies
1332+
// when stop is initiated by cancelling the context passed to Start.
12651333
//
1266-
// There's no need to call this method if a hard stop has already been initiated
1267-
// by cancelling the context passed to Start or by calling StopAndCancel.
1334+
// There's no need to call this method if shutdown has already been initiated by
1335+
// cancelling the context passed to Start or by calling StopAndCancel.
12681336
func (c *Client[TTx]) Stop(ctx context.Context) error {
12691337
shouldStop, stopped, finalizeStop := c.baseStartStop.StopInit()
12701338
if !shouldStop {
@@ -1283,10 +1351,11 @@ func (c *Client[TTx]) Stop(ctx context.Context) error {
12831351

12841352
// StopAndCancel shuts down the client and cancels all work in progress. It is a
12851353
// more aggressive stop than Stop because the contexts for any in-progress jobs
1286-
// are cancelled. However, it still waits for jobs to complete before returning,
1287-
// even though their contexts are cancelled. If the provided context is done
1288-
// before shutdown has completed, StopAndCancel will return immediately with the
1289-
// context's error.
1354+
// are cancelled immediately. If StopAbandonTimeout is configured, jobs that
1355+
// still remain running after the timeout are abandoned; otherwise, StopAndCancel
1356+
// waits for jobs to complete even though their contexts are cancelled. If the
1357+
// provided context is done before shutdown has completed, StopAndCancel will
1358+
// return immediately with the context's error.
12901359
//
12911360
// This can also be initiated by cancelling the context passed to Start. There is
12921361
// no need to call this method if the context passed to Start is cancelled
@@ -1298,7 +1367,7 @@ func (c *Client[TTx]) Stop(ctx context.Context) error {
12981367
// graceful stop semantics without requiring manual orchestration of Stop and
12991368
// StopAndCancel.
13001369
func (c *Client[TTx]) StopAndCancel(ctx context.Context) error {
1301-
c.baseService.Logger.InfoContext(ctx, c.baseService.Name+": Hard stop started; cancelling all work")
1370+
c.baseService.Logger.InfoContext(ctx, c.baseService.Name+": Soft stop started; cancelling all work")
13021371
c.workCancel(rivercommon.ErrStop)
13031372

13041373
shouldStop, stopped, finalizeStop := c.baseStartStop.StopInit()
@@ -2298,6 +2367,7 @@ func (c *Client[TTx]) producerAdd(queueName string, queueConfig QueueConfig) (*p
22982367
JobStuckCount: &c.stuckJobCount,
22992368
JobStuckThreshold: c.config.JobStuckThreshold,
23002369
JobTimeout: c.config.JobTimeout,
2370+
JobUpdateCallback: c.subscriptionManager.distributeJobUpdates,
23012371
MaxWorkers: queueConfig.MaxWorkers,
23022372
Notifier: c.notifier,
23032373
Queue: queueName,

0 commit comments

Comments
 (0)