| 9310 | } |
| 9311 | |
| 9312 | func (p *Server) recoverStaleChats(ctx context.Context) { |
| 9313 | staleAfter := p.clock.Now().Add(-p.inFlightChatStaleAfter) |
| 9314 | staleChats, err := p.db.GetStaleChats(ctx, staleAfter) |
| 9315 | if err != nil { |
| 9316 | p.logger.Error(ctx, "failed to get stale chats", slog.Error(err)) |
| 9317 | return |
| 9318 | } |
| 9319 | |
| 9320 | recovered := 0 |
| 9321 | for _, chat := range staleChats { |
| 9322 | p.logger.Info(ctx, "recovering stale chat", |
| 9323 | slog.F("chat_id", chat.ID), |
| 9324 | slog.F("status", chat.Status)) |
| 9325 | |
| 9326 | // Use a transaction with FOR UPDATE to avoid a TOCTOU race: |
| 9327 | // between GetStaleChats (a bare SELECT) and here, the chat's |
| 9328 | // heartbeat may have been refreshed. We re-check freshness |
| 9329 | // under the row lock before resetting. |
| 9330 | err := p.db.InTx(func(tx database.Store) error { |
| 9331 | locked, lockErr := tx.GetChatByIDForUpdate(ctx, chat.ID) |
| 9332 | if lockErr != nil { |
| 9333 | return xerrors.Errorf("lock chat for recovery: %w", lockErr) |
| 9334 | } |
| 9335 | |
| 9336 | switch locked.Status { |
| 9337 | case database.ChatStatusRunning: |
| 9338 | // Re-check: only recover if the chat is still stale. |
| 9339 | // A valid heartbeat at or after the threshold means |
| 9340 | // the chat was refreshed after our snapshot. |
| 9341 | if locked.HeartbeatAt.Valid && !locked.HeartbeatAt.Time.Before(staleAfter) { |
| 9342 | p.logger.Debug(ctx, "chat heartbeat refreshed since snapshot, skipping recovery", |
| 9343 | slog.F("chat_id", chat.ID)) |
| 9344 | return nil |
| 9345 | } |
| 9346 | case database.ChatStatusRequiresAction: |
| 9347 | // Re-check: the chat may have been updated after |
| 9348 | // our snapshot, similar to the heartbeat check for |
| 9349 | // running chats. |
| 9350 | if !locked.UpdatedAt.Before(staleAfter) { |
| 9351 | p.logger.Debug(ctx, "chat updated since snapshot, skipping recovery", |
| 9352 | slog.F("chat_id", chat.ID)) |
| 9353 | return nil |
| 9354 | } |
| 9355 | case database.ChatStatusWaiting: |
| 9356 | // Deferred-promote stranding: worker died before its |
| 9357 | // post-cancel cleanup ran. Re-check freshness. |
| 9358 | if !locked.UpdatedAt.Before(staleAfter) { |
| 9359 | p.logger.Debug(ctx, "chat updated since snapshot, skipping recovery", |
| 9360 | slog.F("chat_id", chat.ID)) |
| 9361 | return nil |
| 9362 | } |
| 9363 | default: |
| 9364 | // Status changed since our snapshot; skip. |
| 9365 | p.logger.Debug(ctx, "chat status changed since snapshot, skipping recovery", |
| 9366 | slog.F("chat_id", chat.ID), |
| 9367 | slog.F("status", locked.Status)) |
| 9368 | return nil |
| 9369 | } |