| 529 | } |
| 530 | |
| 531 | func (f *logFollower) follow() { |
| 532 | var cancel context.CancelFunc |
| 533 | f.ctx, cancel = context.WithCancel(f.ctx) |
| 534 | defer cancel() |
| 535 | // note that we only need to subscribe to updates if the job is not yet |
| 536 | // complete. |
| 537 | if !f.complete { |
| 538 | subCancel, err := f.pubsub.SubscribeWithErr( |
| 539 | provisionersdk.ProvisionerJobLogsNotifyChannel(f.jobID), |
| 540 | f.listener, |
| 541 | ) |
| 542 | if err != nil { |
| 543 | httpapi.Write(f.ctx, f.rw, http.StatusInternalServerError, codersdk.Response{ |
| 544 | Message: "failed to subscribe to job updates", |
| 545 | Detail: err.Error(), |
| 546 | }) |
| 547 | return |
| 548 | } |
| 549 | defer subCancel() |
| 550 | // Move cancel up the stack so it happens before unsubscribing, |
| 551 | // otherwise we can end up in a deadlock due to how the |
| 552 | // in-memory pubsub does mutex locking on send/unsubscribe. |
| 553 | defer cancel() |
| 554 | |
| 555 | // we were provided `complete` prior to starting this subscription, so |
| 556 | // we also need to check whether the job is now complete, in case the |
| 557 | // job completed between the last time we queried the job and the start |
| 558 | // of the subscription. If the job completes after this, we will get |
| 559 | // a notification on the subscription. |
| 560 | job, err := f.db.GetProvisionerJobByID(f.ctx, f.jobID) |
| 561 | if err != nil { |
| 562 | httpapi.Write(f.ctx, f.rw, http.StatusInternalServerError, codersdk.Response{ |
| 563 | Message: "failed to query job", |
| 564 | Detail: err.Error(), |
| 565 | }) |
| 566 | return |
| 567 | } |
| 568 | f.complete = jobIsComplete(f.logger, job) |
| 569 | f.logger.Debug(f.ctx, "queried job after subscribe", slog.F("complete", f.complete)) |
| 570 | } |
| 571 | |
| 572 | var err error |
| 573 | f.conn, err = websocket.Accept(f.rw, f.r, nil) |
| 574 | if err != nil { |
| 575 | httpapi.Write(f.ctx, f.rw, http.StatusBadRequest, codersdk.Response{ |
| 576 | Message: "Failed to accept websocket.", |
| 577 | Detail: err.Error(), |
| 578 | }) |
| 579 | return |
| 580 | } |
| 581 | defer f.conn.Close(websocket.StatusNormalClosure, "done") |
| 582 | go httpapi.HeartbeatClose(f.ctx, f.logger, cancel, f.conn) |
| 583 | f.enc = wsjson.NewEncoder[codersdk.ProvisionerJobLog](f.conn, websocket.MessageText) |
| 584 | |
| 585 | // query for logs once right away, so we can get historical data from before |
| 586 | // subscription |
| 587 | if err := f.query(); err != nil { |
| 588 | if f.ctx.Err() == nil && !xerrors.Is(err, io.EOF) { |