| 92 | } |
| 93 | |
| 94 | func (w *Watchdog) subscribeMonitor() { |
| 95 | defer w.wg.Done() |
| 96 | tmr := w.clock.NewTimer(periodTimeout) |
| 97 | defer tmr.Stop("subscribe") |
| 98 | beats := make(chan struct{}) |
| 99 | unsub, err := w.ps.Subscribe(EventPubsubWatchdog, func(context.Context, []byte) { |
| 100 | w.logger.Debug(w.ctx, "got heartbeat for pubsub watchdog") |
| 101 | select { |
| 102 | case <-w.ctx.Done(): |
| 103 | case beats <- struct{}{}: |
| 104 | } |
| 105 | }) |
| 106 | if err != nil { |
| 107 | w.logger.Critical(w.ctx, "watchdog failed to subscribe", slog.Error(err)) |
| 108 | close(w.timeout) |
| 109 | return |
| 110 | } |
| 111 | defer unsub() |
| 112 | for { |
| 113 | select { |
| 114 | case <-w.ctx.Done(): |
| 115 | w.logger.Debug(w.ctx, "context done; exiting subscribeMonitor") |
| 116 | return |
| 117 | case <-beats: |
| 118 | // c.f. https://pkg.go.dev/time#Timer.Reset |
| 119 | if !tmr.Stop() { |
| 120 | <-tmr.C |
| 121 | } |
| 122 | tmr.Reset(periodTimeout) |
| 123 | case <-tmr.C: |
| 124 | buf := new(strings.Builder) |
| 125 | _ = pprof.Lookup("goroutine").WriteTo(buf, 1) |
| 126 | w.logger.Critical(w.ctx, "pubsub watchdog timeout", slog.F("goroutines", buf.String())) |
| 127 | close(w.timeout) |
| 128 | return |
| 129 | } |
| 130 | } |
| 131 | } |