Method JobSchedule

riverdriver/riversqlite/river_sqlite_driver.go:728–833

(ctx context.Context, params *riverdriver.JobScheduleParams)

}

func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobScheduleParams) ([]*riverdriver.JobScheduleResult, error) {
	// This operation diverges the most from the Postgres version out of all the
	// others by far. The Postgres version is one gigantic query that can't
	// reproduce here because (1) SQLite doesn't support `UPDATE` in CTEs, and
	// (2) sqlc's handling around any kind of bulk upsert mechanisms (i.e. via
	// arrays, which don't exist, or jsonb) is completely non-functional.
	//
	// Instead, we break the operation up into four much smaller queries. Each
	// query is mostly trivial, so this does have the effect of simplifying
	// things conceptually by quite a lot. It is more Go code though, and one
	// could argue that it's not as concurrently safe as the single query
	// version (i.e. an advisory lock should maybe be taken around when checking
	// rows to avoid non-repeatable read anomalies), but since only one writer
	// at a time is possible for SQLite, this should be okay.
	return dbutil.WithTxV(ctx, e, func(ctx context.Context, execTx riverdriver.ExecutorTx) ([]*riverdriver.JobScheduleResult, error) {
		ctx = schemaTemplateParam(ctx, params.Schema)
		dbtx := templateReplaceWrapper{dbtx: e.driver.UnwrapTx(execTx), replacer: &e.driver.replacer}

		eligibleJobs, err := dbsqlc.New().JobScheduleGetEligible(schemaTemplateParam(ctx, params.Schema), dbtx, &dbsqlc.JobScheduleGetEligibleParams{
			Max: int64(params.Max),
			Now: timeStringNullable(params.Now),
		})
		if err != nil {
			return nil, interpretError(err)
		}

		var (
			discardIDs      []int64
			nonUniqueIDs    []int64
			scheduledResMap = make(map[int64]*riverdriver.JobScheduleResult)
		)

		for _, eligibleJob := range eligibleJobs {
			if eligibleJob.UniqueKey == nil {
				nonUniqueIDs = append(nonUniqueIDs, eligibleJob.ID)
				continue
			}

			internal, err := dbsqlc.New().JobScheduleGetCollision(ctx, dbtx, &dbsqlc.JobScheduleGetCollisionParams{
				ID:        eligibleJob.ID,
				UniqueKey: eligibleJob.UniqueKey,
			})
			if err != nil && !errors.Is(err, sql.ErrNoRows) {
				return nil, interpretError(err)
			}

			if internal.ID != 0 {
				discardIDs = append(discardIDs, eligibleJob.ID)
				continue
			}

			// We must set available immediately rather than in the batch below
			// because it's possible for eligible jobs to be unique conflicting
			// with each other. So when we start this loop two jobs could be
			// scheduled with no unique collision, but after the first one is
			// scheduled, then the second one no longer can and is discarded.
			updatedJobs, err := dbsqlc.New().JobScheduleSetAvailable(ctx, dbtx, []int64{eligibleJob.ID})
			if err != nil {