(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow)
| 127 | } |
| 128 | |
| 129 | func (w *Worker[T, TTx]) workJob(ctx context.Context, tb testing.TB, tx TTx, job *rivertype.JobRow) (*WorkResult, error) { |
| 130 | tb.Helper() |
| 131 | |
| 132 | timeGen := w.config.Test.Time |
| 133 | if timeGen == nil { |
| 134 | timeGen = &baseservice.UnStubbableTimeGenerator{} |
| 135 | } |
| 136 | |
| 137 | exec := w.client.Driver().UnwrapExecutor(tx) |
| 138 | subscribeCh := make(chan []jobcompleter.CompleterJobUpdated, 1) |
| 139 | archetype := riversharedtest.BaseServiceArchetype(tb) |
| 140 | if w.config.Logger != nil { |
| 141 | archetype.Logger = w.config.Logger |
| 142 | } |
| 143 | if withStub, ok := timeGen.(baseservice.TimeGeneratorWithStub); ok { |
| 144 | archetype.Time = withStub |
| 145 | } else { |
| 146 | archetype.Time = &baseservice.TimeGeneratorWithStubWrapper{TimeGenerator: timeGen} |
| 147 | } |
| 148 | completer := jobcompleter.NewInlineCompleter(archetype, w.config.Schema, exec, w.client.Pilot(), subscribeCh) |
| 149 | |
| 150 | for _, hook := range w.config.Hooks { |
| 151 | if withBaseService, ok := hook.(baseservice.WithBaseService); ok { |
| 152 | baseservice.Init(archetype, withBaseService) |
| 153 | } |
| 154 | } |
| 155 | for _, middleware := range w.config.Middleware { |
| 156 | if withBaseService, ok := middleware.(baseservice.WithBaseService); ok { |
| 157 | baseservice.Init(archetype, withBaseService) |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | updatedJobRow, err := exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{ |
| 162 | ID: job.ID, |
| 163 | Attempt: job.Attempt + 1, |
| 164 | AttemptDoUpdate: true, |
| 165 | AttemptedAt: ptrutil.Ptr(timeGen.Now()), |
| 166 | AttemptedAtDoUpdate: true, |
| 167 | AttemptedBy: append(job.AttemptedBy, w.config.ID), |
| 168 | AttemptedByDoUpdate: true, |
| 169 | Schema: w.config.Schema, |
| 170 | StateDoUpdate: true, |
| 171 | State: rivertype.JobStateRunning, |
| 172 | }) |
| 173 | if err != nil && !errors.Is(err, rivertype.ErrNotFound) { |
| 174 | return nil, fmt.Errorf("test worker internal error: failed to update job to running state: %w", err) |
| 175 | } |
| 176 | job = updatedJobRow |
| 177 | |
| 178 | workUnit := (&workUnitFactoryWrapper[T]{worker: w.worker}).MakeUnit(job) |
| 179 | |
| 180 | // populate river client into context: |
| 181 | ctx = WorkContext(ctx, w.client) |
| 182 | // TODO: remove ContextKeyInsideTestWorker |
| 183 | ctx = context.WithValue(ctx, execution.ContextKeyInsideTestWorker{}, true) |
| 184 | |
| 185 | // jobCancel will always be called by the executor to prevent leaks. |
| 186 | jobCtx, jobCancel := context.WithCancelCause(ctx) |
no test coverage detected