launchHeartbeat starts a goroutine that periodically calls TouchStep to keep the step and run rows alive during long-running streams. The goroutine also listens on the service's threshold-change channel so that a runtime SetStaleAfter call immediately resets the ticker instead of waiting for the ol
(ctx context.Context, svc *Service, stepID, runID, chatID uuid.UUID, done <-chan struct{})
| 357 | // instead of waiting for the old (possibly longer) period to elapse. |
| 358 | // The goroutine exits when done is closed or ctx is canceled. |
| 359 | func launchHeartbeat(ctx context.Context, svc *Service, stepID, runID, chatID uuid.UUID, done <-chan struct{}) { |
| 360 | if svc == nil { |
| 361 | return |
| 362 | } |
| 363 | go func() { |
| 364 | // Subscribe before reading the interval. The channel invalidates |
| 365 | // the interval, so any concurrent SetStaleAfter either happened |
| 366 | // before this interval read or will close thresholdCh below. |
| 367 | thresholdCh := svc.thresholdChan() |
| 368 | interval := svc.heartbeatInterval() |
| 369 | ticker := svc.clock.NewTicker(interval, "chatdebug", "heartbeat") |
| 370 | defer ticker.Stop() |
| 371 | resetTicker := func() { |
| 372 | if newInterval := svc.heartbeatInterval(); newInterval != interval { |
| 373 | interval = newInterval |
| 374 | ticker.Reset(interval, "chatdebug", "heartbeat") |
| 375 | } |
| 376 | } |
| 377 | |
| 378 | for { |
| 379 | select { |
| 380 | case <-ctx.Done(): |
| 381 | return |
| 382 | case <-done: |
| 383 | return |
| 384 | case <-thresholdCh: |
| 385 | // SetStaleAfter was called; re-read the interval |
| 386 | // and reset the ticker immediately. |
| 387 | thresholdCh = svc.thresholdChan() |
| 388 | resetTicker() |
| 389 | case <-ticker.C: |
| 390 | if err := svc.TouchStep(ctx, stepID, runID, chatID); err != nil { |
| 391 | svc.log.Debug(ctx, "heartbeat touch failed", |
| 392 | slog.Error(err), |
| 393 | slog.F("step_id", stepID), |
| 394 | ) |
| 395 | } |
| 396 | // Also re-read interval on every tick as a |
| 397 | // secondary check. |
| 398 | resetTicker() |
| 399 | } |
| 400 | } |
| 401 | }() |
| 402 | } |
| 403 | |
| 404 | func wrapStreamSeq( |
| 405 | ctx context.Context, |