Method distributeJobEvent

subscription_manager.go:155–196

Distribute a single event into any listening subscriber channels. Job events should specify the job and stats, while queue events should only specify the queue. MUST be called with sm.mu already held.

(ctx context.Context, job *rivertype.JobRow, stats *JobStatistics, snoozed bool)

//
// MUST be called with sm.mu already held.
func (sm *subscriptionManager) distributeJobEvent(ctx context.Context, job *rivertype.JobRow, stats *JobStatistics, snoozed bool) {
	var event *Event
	if snoozed {
		event = &Event{Kind: EventKindJobSnoozed, Job: job, JobStats: stats}
	} else {
		switch job.State {
		case rivertype.JobStateCancelled:
			event = &Event{Kind: EventKindJobCancelled, Job: job, JobStats: stats}
		case rivertype.JobStateCompleted:
			event = &Event{Kind: EventKindJobCompleted, Job: job, JobStats: stats}
		case rivertype.JobStateAvailable, rivertype.JobStateDiscarded, rivertype.JobStateRetryable, rivertype.JobStateRunning:
			event = &Event{Kind: EventKindJobFailed, Job: job, JobStats: stats}
		case rivertype.JobStatePending, rivertype.JobStateScheduled:
			// job state may be set to scheduled, but only for snoozed jobs, so
			// the case at the top should always take precedence before this
			panic(fmt.Sprintf("completion subscriber unexpectedly received job in %s state, river bug", job.State))
		default:
			// linter exhaustive rule prevents this from being reached
			panic("unreachable state to distribute, river bug")
		}
	}

	// All subscription channels are non-blocking so this is always fast and
	// there's no risk of falling behind what producers are sending.
	for _, sub := range sm.subscriptions {
		if sub.ListensFor(event.Kind) {
			// TODO: THIS IS UNSAFE AND WILL LEAD TO DROPPED EVENTS.
			//
			// We are allocating subscriber channels with a fixed size of 1000, but
			// potentially processing job events in batches of 5000 (batch completer
			// max batch size). It's probably not possible for the subscriber to keep
			// up with these bursts.
			select {
			case sub.Chan <- event:
			default:
				sm.Logger.WarnContext(ctx, sm.Name+": Subscription event dropped due to full buffer",
					slog.String("event_kind", string(event.Kind)),
				)
			}
		}
	}
}

func (sm *subscriptionManager) distributeQueueEvent(event *Event) {
	sm.distributeQueueEventWithContext(context.Background(), event)