| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strconv" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/activator" |
| 14 | "github.com/carlosframework/platform/internal/hostclaim" |
| 15 | "github.com/carlosframework/platform/internal/instancectl" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/runs" |
| 19 | ) |
| 20 | |
| 21 | // runPollRetry paces a failed long-poll; runRecheckEvery re-reads the |
| 22 | // record while a run executes, a backstop for a cancel whose serial bump |
| 23 | // this box missed. |
| 24 | var ( |
| 25 | runPollRetry = time.Second |
| 26 | runRecheckEvery = 30 * time.Second |
| 27 | // runInterruptEvery is how often follow asks the handle whether an |
| 28 | // instance stop or a retired generation interrupted the run, and |
| 29 | // runInterruptGrace how long it then waits for the unit's own done |
| 30 | // status, which wins if the command had already finished (re-review 6). |
| 31 | runInterruptEvery = 250 * time.Millisecond |
| 32 | runInterruptGrace = 2 * time.Second |
| 33 | ) |
| 34 | |
| 35 | // runOutputGrace bounds the output once the command has ended: the unit's |
| 36 | // stop (whose streams drain into the chunker) and the last chunk. The |
| 37 | // chunker's context outlives the command, so a cancel or a timeout never |
| 38 | // gives up the last chunk; this bound is what keeps a store that refuses |
| 39 | // every Put from retrying under the chunker's lock forever. It is longer |
| 40 | // than runStopBound. |
| 41 | var runOutputGrace = 5 * time.Minute |
| 42 | |
| 43 | // runFinishBound bounds the retries of result.json and of the done CAS. |
| 44 | var runFinishBound = 2 * time.Minute |
| 45 | |
| 46 | const ( |
| 47 | // runStopBound bounds CmdHandle.Stop: above the activator's own stop |
| 48 | // bound, which covers the start, the stop, the cgroup and both streams. |
| 49 | runStopBound = 3 * time.Minute |
| 50 | // runReadBound bounds one read of the record. |
| 51 | runReadBound = 30 * time.Second |
| 52 | ) |
| 53 | |
| 54 | var ( |
| 55 | errRunCancelledBeforeLaunch = errors.New("host: the run was cancelled before launch") |
| 56 | errRunImportAtLaunch = errors.New("host: the instance is an import that is not committed") |
| 57 | errRunIncarnation = errors.New("host: the record is another incarnation") |
| 58 | errRunNotOurs = errors.New("host: the record no longer names this run as this box's") |
| 59 | ) |
| 60 | |
| 61 | // exitOf reads systemd's ExecMainCode and ExecMainStatus. `systemctl show` |
| 62 | // prints the code as a number (1 exited, 2 killed, 3 dumped); the words |
| 63 | // are accepted too. A signal leaves exit 0 and names the signal. |
| 64 | func exitOf(st instancectl.CmdStatus) (int, string) { |
| 65 | n, _ := strconv.Atoi(st.Status) |
| 66 | switch st.Code { |
| 67 | case "2", "3", "killed", "dumped": |
| 68 | return 0, strconv.Itoa(n) |
| 69 | } |
| 70 | return n, "" |
| 71 | } |
| 72 | |
| 73 | // runResultForLaunch maps RunCmd's refusals, and Prepare's, to the result |
| 74 | // the member reads. A route that is not an instance's (a worker) is |
| 75 | // refused as a flat box is: runs need an instance on the nested layout. |
| 76 | // Anything else (a busy hold, a failed download, a failed start) is the |
| 77 | // box failing to start the command. |
| 78 | func runResultForLaunch(err error) runs.Result { |
| 79 | var we *activator.CmdWakeError |
| 80 | var se *activator.CmdStoppedError |
| 81 | switch { |
| 82 | case errors.Is(err, activator.ErrCmdClaim): |
| 83 | return runs.Result{Refused: runs.RefusedClaim} |
| 84 | case errors.Is(err, activator.ErrCmdNotNested), errors.Is(err, activator.ErrCmdNotInstance): |
| 85 | return runs.Result{Refused: runs.RefusedFlat} |
| 86 | case errors.As(err, &we): |
| 87 | return runs.Result{Refused: runs.RefusedWake, Detail: we.Reason} |
| 88 | case errors.As(err, &se): |
| 89 | return runs.Result{Interrupted: se.Reason} |
| 90 | case errors.Is(err, errRunCancelledBeforeLaunch): |
| 91 | return runs.Result{Cancelled: true} |
| 92 | case errors.Is(err, errRunImportAtLaunch): |
| 93 | return runs.Result{Refused: runs.RefusedImport} |
| 94 | case errors.Is(err, errRunIncarnation): |
| 95 | return runs.Result{Refused: runs.RefusedIncarnation} |
| 96 | } |
| 97 | return runs.Result{Interrupted: runs.InterruptedStartFailed} |
| 98 | } |
| 99 | |
| 100 | // recordIncarnation is the incarnation request.json must carry for rec; |
| 101 | // "" when it cannot be computed, which matches no request. |
| 102 | func recordIncarnation(rec provision.Record) string { |
| 103 | inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt) |
| 104 | if err != nil { |
| 105 | return "" |
| 106 | } |
| 107 | return inc |
| 108 | } |
| 109 | |
| 110 | // runFinishOurs is RunFinish for a run the record names as picked by box |
| 111 | // under boot. RunFinish itself does not check who picked the run. |
| 112 | func runFinishOurs(id, box, boot string) func(*provision.Record) error { |
| 113 | finish := provision.RunFinish(id) |
| 114 | return func(rec *provision.Record) error { |
| 115 | if r := rec.ActiveRun; r == nil || r.ID != id || r.Box != box || r.BootID != boot { |
| 116 | return fmt.Errorf("%w: the record does not name run %s as this box's", provision.ErrRunConflict, id) |
| 117 | } |
| 118 | return finish(rec) |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | // execute is one picked run's life (run spec §5.2 to §5.7): the |
| 123 | // incarnation, the launch (Prepare checks the record again and stages the |
| 124 | // files), the poll, a cancel or a timeout, then the unit stopped, every |
| 125 | // chunk written, result.json, the done CAS, the run directory deleted and |
| 126 | // the hold released, in that order. |
| 127 | func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) { |
| 128 | defer w.forget(req.ID) |
| 129 | instDir := provision.InstanceDir(w.dataDir, rec.Host) |
| 130 | if req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host || req.Incarnation == "" || req.Incarnation != recordIncarnation(rec) { |
| 131 | w.finish(ctx, rec, req, runs.Result{Refused: runs.RefusedIncarnation}, nil) |
| 132 | return |
| 133 | } |
| 134 | // The chunker's context is not the run's: a cancel or a timeout must |
| 135 | // not give up the last chunk. endOutput is armed once the command has |
| 136 | // ended (runOutputGrace). |
| 137 | octx, endOutput := context.WithCancel(context.WithoutCancel(ctx)) |
| 138 | defer endOutput() |
| 139 | chunks := runs.NewChunker(octx, w.store, rec.Account, rec.App, req.ID) |
| 140 | closeOutput := func(res *runs.Result) { |
| 141 | last, truncated, err := chunks.Close() |
| 142 | if err != nil { |
| 143 | log.Printf("agent: runs: %s %s: output: %v", rec.Host, req.ID, err) |
| 144 | } |
| 145 | res.LastSeq, res.Truncated = last, truncated |
| 146 | } |
| 147 | // notOurs is prepare's own verdict: RunCmd reports a stop that |
| 148 | // overlapped Prepare in place of Prepare's error, and a run the record |
| 149 | // no longer gives this box must get no result whatever RunCmd says. |
| 150 | notOurs := false |
| 151 | // prepRefused: Prepare answered an error, so its own result, not the |
| 152 | // launch context's end, is the run's story (a staging that meets the |
| 153 | // deadline stays start failed). recheckErr is the post-switch |
| 154 | // Recheck's answer: a refusal keeps its own result too, but a record |
| 155 | // read the run's own cancel or deadline cut short is not a refusal |
| 156 | // (re-review 7). |
| 157 | prepRefused := false |
| 158 | var recheckErr error |
| 159 | spec := activator.CmdSpec{ |
| 160 | RunID: req.ID, |
| 161 | Args: req.Argv, |
| 162 | RunDir: filepath.Join(instDir, runs.DirName, req.ID), |
| 163 | Stdout: chunks.Stream(runs.StreamOut), |
| 164 | Stderr: chunks.Stream(runs.StreamErr), |
| 165 | Prepare: func(ctx context.Context) error { |
| 166 | err := w.prepare(ctx, instDir, req, h, launchDeadline(rec, req, w.now())) |
| 167 | notOurs, prepRefused = errors.Is(err, errRunNotOurs), err != nil |
| 168 | return err |
| 169 | }, |
| 170 | Recheck: func(ctx context.Context) error { |
| 171 | err := w.launchable(ctx, req, h) |
| 172 | notOurs, recheckErr = errors.Is(err, errRunNotOurs), err |
| 173 | return err |
| 174 | }, |
| 175 | } |
| 176 | // The launch runs under the run's deadline and its cancel, the whole |
| 177 | // admission through (re-review 5): the wake, Prepare, a wait for a |
| 178 | // swap's switch and the last checks. A run cancelled or expired at |
| 179 | // any of them starts nothing. |
| 180 | lctx, lcancel := w.launchContext(ctx, rec, req, h) |
| 181 | handle, err := w.launch.RunCmd(lctx, rt, spec) |
| 182 | launchEnded := lctx.Err() |
| 183 | lcancel() |
| 184 | if err != nil { |
| 185 | bound := time.AfterFunc(runOutputGrace, endOutput) |
| 186 | defer bound.Stop() |
| 187 | if notOurs || errors.Is(err, errRunNotOurs) { |
| 188 | // The record gives the run to no one, or to another boot: |
| 189 | // whoever holds it writes its result. |
| 190 | chunks.Close() |
| 191 | w.cleanUp(rec, req, nil) |
| 192 | log.Printf("agent: runs: %s %s: not launched: %v", rec.Host, req.ID, err) |
| 193 | return |
| 194 | } |
| 195 | res := runResultForLaunch(err) |
| 196 | recheckRefused := recheckErr != nil && !(launchEnded != nil && endedByContext(recheckErr)) |
| 197 | if launchEnded != nil && !prepRefused && !recheckRefused { |
| 198 | // The admission ended with the run's own cancel or deadline: |
| 199 | // the run's own result, as after launch (re-review 5). |
| 200 | cancelled := cancelRequested(h) |
| 201 | res = runs.Result{Cancelled: cancelled, TimedOut: !cancelled} |
| 202 | } |
| 203 | if res.Interrupted == runs.InterruptedStartFailed { |
| 204 | log.Printf("agent: runs: %s %s: start: %v", rec.Host, req.ID, err) |
| 205 | } |
| 206 | closeOutput(&res) |
| 207 | w.finish(ctx, rec, req, res, nil) |
| 208 | return |
| 209 | } |
| 210 | log.Printf("agent: runs: %s: run %s started (%s)", rec.Host, req.ID, handle.Invocation()) |
| 211 | w.runEvent(rec, runStartedLine(req)) |
| 212 | res := runs.Result{Started: w.now().UTC()} |
| 213 | status, doneAt := w.follow(ctx, rec, req, handle, h, &res) |
| 214 | bound := time.AfterFunc(runOutputGrace, endOutput) |
| 215 | defer bound.Stop() |
| 216 | // The unit is stopped and settled after every ending, done included |
| 217 | // (run spec §5.4), so its streams have reached their end before the |
| 218 | // chunker closes and the result names the last chunk. |
| 219 | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runStopBound) |
| 220 | if err := handle.Stop(sctx); err != nil { |
| 221 | log.Printf("agent: runs: %s %s: stop: %v", rec.Host, req.ID, err) |
| 222 | } |
| 223 | cancel() |
| 224 | if status == nil && (res.Cancelled || res.TimedOut) { |
| 225 | pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineCmdPoll) |
| 226 | if st, err := handle.Poll(pctx); err == nil && st.State == instancectl.CmdDone { |
| 227 | status = &st |
| 228 | } |
| 229 | pcancel() |
| 230 | } |
| 231 | if status != nil && status.State == instancectl.CmdDone { |
| 232 | res.Exit, res.Signal = exitOf(*status) |
| 233 | } |
| 234 | reason, askedAt := handle.Interrupted() |
| 235 | if reason != "" && status != nil && status.State == instancectl.CmdDone && !doneAt.IsZero() && askedAt.After(doneAt) { |
| 236 | // The instance stop was asked for after the done status arrived: |
| 237 | // the command had finished on its own, and the done status wins. |
| 238 | reason = "" |
| 239 | } |
| 240 | if reason != "" { |
| 241 | res.Interrupted = reason |
| 242 | } else if status != nil && status.State == instancectl.CmdGone && !res.Cancelled && !res.TimedOut { |
| 243 | res.Interrupted = runs.InterruptedGone |
| 244 | } |
| 245 | closeOutput(&res) |
| 246 | w.finish(ctx, rec, req, res, handle) |
| 247 | } |
| 248 | |
| 249 | // endedByContext reports whether err is a context's end: a read cut short |
| 250 | // by a cancel or a deadline, never a refusal on the record's say. |
| 251 | func endedByContext(err error) bool { |
| 252 | return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) |
| 253 | } |
| 254 | |
| 255 | // launchContext is ctx bounded by the run's deadline and ended by its |
| 256 | // cancel: what RunCmd runs under (re-review 5). The cancel func must be |
| 257 | // called once RunCmd has returned. |
| 258 | func (w *runsWorker) launchContext(ctx context.Context, rec provision.Record, req runs.Request, h *runHandle) (context.Context, context.CancelFunc) { |
| 259 | lctx, cancel := context.WithDeadline(ctx, launchDeadline(rec, req, w.now())) |
| 260 | go func() { |
| 261 | select { |
| 262 | case <-h.cancel: |
| 263 | cancel() |
| 264 | case <-lctx.Done(): |
| 265 | } |
| 266 | }() |
| 267 | return lctx, cancel |
| 268 | } |
| 269 | |
| 270 | // launchDeadline is the run's deadline from the record (RunPick set it to |
| 271 | // the pickup plus the timeout), or now plus the timeout without one. |
| 272 | func launchDeadline(rec provision.Record, req runs.Request, now time.Time) time.Time { |
| 273 | if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() { |
| 274 | return r.Deadline |
| 275 | } |
| 276 | return now.Add(req.Timeout()) |
| 277 | } |
| 278 | |
| 279 | // prepare is the run's CmdSpec.Prepare, run under the activator's hold |
| 280 | // right before the unit starts (run spec §5.2): the record must still name |
| 281 | // the run as this box's under this boot, not cancelled, not an uncommitted |
| 282 | // import, the same incarnation. Then the files are staged, a cancel cutting |
| 283 | // the download short, and the record is checked once more, so a cancel |
| 284 | // that lands while the files download still starts no unit. All of it ends |
| 285 | // at the run's deadline: a stalled download must not keep the hold (and |
| 286 | // with it every later run on the instance) forever. |
| 287 | func (w *runsWorker) prepare(ctx context.Context, instDir string, req runs.Request, h *runHandle, deadline time.Time) error { |
| 288 | ctx, cancel := context.WithDeadline(ctx, deadline) |
| 289 | defer cancel() |
| 290 | if err := w.launchable(ctx, req, h); err != nil { |
| 291 | return err |
| 292 | } |
| 293 | sctx, stop := context.WithCancel(ctx) |
| 294 | defer stop() |
| 295 | go func() { |
| 296 | select { |
| 297 | case <-h.cancel: |
| 298 | stop() |
| 299 | case <-sctx.Done(): |
| 300 | } |
| 301 | }() |
| 302 | if _, err := runs.StageFiles(sctx, w.store, instDir, req); err != nil { |
| 303 | if cancelRequested(h) { |
| 304 | return errRunCancelledBeforeLaunch |
| 305 | } |
| 306 | return fmt.Errorf("host: stage the run's files: %w", err) |
| 307 | } |
| 308 | return w.launchable(ctx, req, h) |
| 309 | } |
| 310 | |
| 311 | // launchable reads the record and says why the run must not start, or nil. |
| 312 | func (w *runsWorker) launchable(ctx context.Context, req runs.Request, h *runHandle) error { |
| 313 | if cancelRequested(h) { |
| 314 | return errRunCancelledBeforeLaunch |
| 315 | } |
| 316 | rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| 317 | defer cancel() |
| 318 | cur, err := provision.GetRecord(rctx, w.store, req.Account, req.App, req.Host) |
| 319 | if errors.Is(err, os.ErrNotExist) { |
| 320 | return errRunNotOurs |
| 321 | } |
| 322 | if err != nil { |
| 323 | return fmt.Errorf("host: read the record before launch: %w", err) |
| 324 | } |
| 325 | r := cur.ActiveRun |
| 326 | switch { |
| 327 | case r == nil || r.ID != req.ID || r.Box != w.box || r.BootID != w.boot || (r.State != provision.RunPicked && r.State != provision.RunCancel): |
| 328 | return errRunNotOurs |
| 329 | case r.State == provision.RunCancel: |
| 330 | return errRunCancelledBeforeLaunch |
| 331 | case cur.ImportUncommitted(): |
| 332 | return errRunImportAtLaunch |
| 333 | case recordIncarnation(*cur) != req.Incarnation: |
| 334 | return errRunIncarnation |
| 335 | } |
| 336 | return nil |
| 337 | } |
| 338 | |
| 339 | func cancelRequested(h *runHandle) bool { |
| 340 | select { |
| 341 | case <-h.cancel: |
| 342 | return true |
| 343 | default: |
| 344 | return false |
| 345 | } |
| 346 | } |
| 347 | |
| 348 | // follow long-polls the unit until it is done or gone, the run is |
| 349 | // cancelled, or its timeout passes. It returns the final status when the |
| 350 | // unit reported one, with when the poll answered it, and marks a cancel or |
| 351 | // a timeout on res. |
| 352 | func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.Request, handle activator.CmdHandle, h *runHandle, res *runs.Result) (*instancectl.CmdStatus, time.Time) { |
| 353 | pctx, stopPolling := context.WithCancel(ctx) |
| 354 | defer stopPolling() |
| 355 | type polled struct { |
| 356 | st instancectl.CmdStatus |
| 357 | at time.Time |
| 358 | } |
| 359 | statuses := make(chan polled, 1) |
| 360 | go func() { |
| 361 | for { |
| 362 | st, err := handle.Poll(pctx) |
| 363 | if pctx.Err() != nil { |
| 364 | return |
| 365 | } |
| 366 | if err != nil { |
| 367 | select { |
| 368 | case <-pctx.Done(): |
| 369 | return |
| 370 | case <-time.After(runPollRetry): |
| 371 | } |
| 372 | continue |
| 373 | } |
| 374 | if st.State != instancectl.CmdRunning { |
| 375 | statuses <- polled{st, w.now()} |
| 376 | return |
| 377 | } |
| 378 | } |
| 379 | }() |
| 380 | // The timeout is the run's own deadline, set at pickup: staging that |
| 381 | // ran long leaves the command less, not a full timeout from launch. |
| 382 | limit := req.Timeout() |
| 383 | if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() { |
| 384 | limit = min(limit, r.Deadline.Sub(w.now())) |
| 385 | } |
| 386 | timeout := time.NewTimer(limit) |
| 387 | defer timeout.Stop() |
| 388 | recheck := time.NewTicker(runRecheckEvery) |
| 389 | defer recheck.Stop() |
| 390 | interrupts := time.NewTicker(runInterruptEvery) |
| 391 | defer interrupts.Stop() |
| 392 | for { |
| 393 | select { |
| 394 | case p := <-statuses: |
| 395 | return &p.st, p.at |
| 396 | case <-interrupts.C: |
| 397 | // An interruption ends the run as interrupted (exit 125) at |
| 398 | // once, even while the activator is still retrying the unit's |
| 399 | // stop (re-review 6): the command's result does not wait on a |
| 400 | // stop that may take a while to be proven. A done status that |
| 401 | // arrives within the grace is still read, so a command that had |
| 402 | // finished first keeps its exit. |
| 403 | if reason, _ := handle.Interrupted(); reason != "" { |
| 404 | grace := time.NewTimer(runInterruptGrace) |
| 405 | defer grace.Stop() |
| 406 | select { |
| 407 | case p := <-statuses: |
| 408 | return &p.st, p.at |
| 409 | case <-grace.C: |
| 410 | return nil, time.Time{} |
| 411 | } |
| 412 | } |
| 413 | case <-h.cancel: |
| 414 | res.Cancelled = true |
| 415 | return nil, time.Time{} |
| 416 | case <-timeout.C: |
| 417 | res.TimedOut = true |
| 418 | return nil, time.Time{} |
| 419 | case <-recheck.C: |
| 420 | rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| 421 | cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host) |
| 422 | cancel() |
| 423 | if err == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == req.ID && cur.ActiveRun.State == provision.RunCancel { |
| 424 | h.requestCancel() |
| 425 | } |
| 426 | } |
| 427 | } |
| 428 | } |
| 429 | |
| 430 | // finish writes the run's result, then CASes the run to done only if the |
| 431 | // result landed and the record names the run as this box's under this |
| 432 | // boot, then cleans up. A result that cannot be written leaves the run |
| 433 | // picked: the sweep ends it as lost with a result of its own, so a run is |
| 434 | // never done without one. |
| 435 | func (w *runsWorker) finish(ctx context.Context, rec provision.Record, req runs.Request, res runs.Result, handle activator.CmdHandle) { |
| 436 | res.Box, res.Ended = w.box, w.now().UTC() |
| 437 | fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runFinishBound) |
| 438 | defer cancel() |
| 439 | err := untilLanded(fctx, func(ctx context.Context) error { |
| 440 | return runs.WriteResult(ctx, w.store, rec.Account, rec.App, req.ID, res) |
| 441 | }) |
| 442 | if err != nil { |
| 443 | log.Printf("agent: runs: %s %s: result: %v", rec.Host, req.ID, err) |
| 444 | } else { |
| 445 | // The ending is logged only once the result has landed (amadan |
| 446 | // #325: run lines are at most once). A result that never landed |
| 447 | // leaves the run picked, and a later agent's closeInterrupted |
| 448 | // writes and logs its own ending; logging this one too would put |
| 449 | // two contradictory endings in the log. A run with no handle never |
| 450 | // had its unit started. |
| 451 | w.runEvent(rec, runEndedLine(req.ID, res, handle != nil)) |
| 452 | if err := untilLanded(fctx, func(ctx context.Context) error { |
| 453 | _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, runFinishOurs(req.ID, w.box, w.boot)) |
| 454 | return err |
| 455 | }); err != nil && !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 456 | log.Printf("agent: runs: %s %s: done: %v", rec.Host, req.ID, err) |
| 457 | } |
| 458 | } |
| 459 | w.cleanUp(rec, req, handle) |
| 460 | } |
| 461 | |
| 462 | // cleanUp deletes the run's directory (an error is logged, never acted |
| 463 | // on) and releases the activator's hold. |
| 464 | func (w *runsWorker) cleanUp(rec provision.Record, req runs.Request, handle activator.CmdHandle) { |
| 465 | if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), req.ID); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 466 | log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, req.ID, err) |
| 467 | } |
| 468 | if handle != nil { |
| 469 | handle.Release() |
| 470 | } |
| 471 | } |
| 472 | |
| 473 | // untilLanded retries f until it succeeds, answers a conflict or a missing |
| 474 | // object (neither changes on a retry), or ctx ends. |
| 475 | func untilLanded(ctx context.Context, f func(context.Context) error) error { |
| 476 | backoff := 250 * time.Millisecond |
| 477 | for { |
| 478 | err := f(ctx) |
| 479 | if err == nil || errors.Is(err, provision.ErrRunConflict) || errors.Is(err, os.ErrNotExist) { |
| 480 | return err |
| 481 | } |
| 482 | select { |
| 483 | case <-ctx.Done(): |
| 484 | return err |
| 485 | case <-time.After(backoff): |
| 486 | } |
| 487 | backoff = min(backoff*2, 5*time.Second) |
| 488 | } |
| 489 | } |
| 490 | |
amadan