AcquireJob acquires a job with one of the given provisioner types and compatible tags from the database. The call blocks until a job is acquired, the context is done, or the database returns an error _other_ than that no jobs are available. If no jobs are available, this method handles retrying as
( ctx context.Context, organization uuid.UUID, worker uuid.UUID, pt []database.ProvisionerType, tags Tags, )
| 89 | // done, or the database returns an error _other_ than that no jobs are available. |
| 90 | // If no jobs are available, this method handles retrying as appropriate. |
| 91 | func (a *Acquirer) AcquireJob( |
| 92 | ctx context.Context, organization uuid.UUID, worker uuid.UUID, pt []database.ProvisionerType, tags Tags, |
| 93 | ) ( |
| 94 | retJob database.ProvisionerJob, retErr error, |
| 95 | ) { |
| 96 | logger := a.logger.With( |
| 97 | slog.F("organization_id", organization), |
| 98 | slog.F("worker_id", worker), |
| 99 | slog.F("provisioner_types", pt), |
| 100 | slog.F("tags", tags)) |
| 101 | logger.Debug(ctx, "acquiring job") |
| 102 | dk := domainKey(organization, pt, tags) |
| 103 | dbTags, err := tags.ToJSON() |
| 104 | if err != nil { |
| 105 | return database.ProvisionerJob{}, err |
| 106 | } |
| 107 | // buffer of 1 so that cancel doesn't deadlock while writing to the channel |
| 108 | clearance := make(chan struct{}, 1) |
| 109 | for { |
| 110 | a.want(organization, pt, tags, clearance) |
| 111 | select { |
| 112 | case <-ctx.Done(): |
| 113 | err := ctx.Err() |
| 114 | logger.Debug(ctx, "acquiring job canceled", slog.Error(err)) |
| 115 | internalError := a.cancel(dk, clearance) |
| 116 | if internalError != nil { |
| 117 | // internalError takes precedence |
| 118 | return database.ProvisionerJob{}, internalError |
| 119 | } |
| 120 | return database.ProvisionerJob{}, err |
| 121 | case <-clearance: |
| 122 | logger.Debug(ctx, "got clearance to call database") |
| 123 | job, err := a.store.AcquireProvisionerJob(ctx, database.AcquireProvisionerJobParams{ |
| 124 | OrganizationID: organization, |
| 125 | StartedAt: sql.NullTime{ |
| 126 | Time: dbtime.Now(), |
| 127 | Valid: true, |
| 128 | }, |
| 129 | WorkerID: uuid.NullUUID{ |
| 130 | UUID: worker, |
| 131 | Valid: true, |
| 132 | }, |
| 133 | Types: pt, |
| 134 | ProvisionerTags: dbTags, |
| 135 | }) |
| 136 | if xerrors.Is(err, sql.ErrNoRows) { |
| 137 | logger.Debug(ctx, "no job available") |
| 138 | continue |
| 139 | } |
| 140 | // we are not going to retry, so signal we are done |
| 141 | internalError := a.done(dk, clearance) |
| 142 | if internalError != nil { |
| 143 | // internal error takes precedence |
| 144 | return database.ProvisionerJob{}, internalError |
| 145 | } |
| 146 | if err != nil { |
| 147 | logger.Warn(ctx, "error attempting to acquire job", slog.Error(err)) |
| 148 | return database.ProvisionerJob{}, xerrors.Errorf("failed to acquire job: %w", err) |