(ctx context.Context)
| 4393 | } |
| 4394 | |
| 4395 | func (p *Server) processOnce(ctx context.Context) { |
| 4396 | if ctx.Err() != nil { |
| 4397 | return |
| 4398 | } |
| 4399 | |
| 4400 | // We detach from the server lifetime to prevent a |
| 4401 | // phantom-acquire race: when the server context is |
| 4402 | // canceled, the pq driver's watchCancel goroutine |
| 4403 | // races with the actual query on the wire. Using a |
| 4404 | // context that cannot be canceled ensures the driver |
| 4405 | // sees the query result if Postgres executed it. |
| 4406 | acquireCtx, acquireCancel := context.WithTimeout( |
| 4407 | context.WithoutCancel(ctx), 10*time.Second, |
| 4408 | ) |
| 4409 | chats, err := p.db.AcquireChats(acquireCtx, database.AcquireChatsParams{ |
| 4410 | StartedAt: time.Now(), |
| 4411 | WorkerID: p.workerID, |
| 4412 | NumChats: p.maxChatsPerAcquire, |
| 4413 | }) |
| 4414 | acquireCancel() |
| 4415 | if err != nil { |
| 4416 | p.logger.Error(ctx, "failed to acquire chats", slog.Error(err)) |
| 4417 | return |
| 4418 | } |
| 4419 | if len(chats) == 0 { |
| 4420 | return |
| 4421 | } |
| 4422 | |
| 4423 | // If the server context was canceled while we were |
| 4424 | // acquiring, release the chats back to pending. |
| 4425 | if ctx.Err() != nil { |
| 4426 | releaseCtx, releaseCancel := context.WithTimeout( |
| 4427 | context.WithoutCancel(ctx), 10*time.Second, |
| 4428 | ) |
| 4429 | for _, chat := range chats { |
| 4430 | _, updateErr := p.db.UpdateChatStatus(releaseCtx, database.UpdateChatStatusParams{ |
| 4431 | ID: chat.ID, |
| 4432 | Status: database.ChatStatusPending, |
| 4433 | WorkerID: uuid.NullUUID{}, |
| 4434 | StartedAt: sql.NullTime{}, |
| 4435 | HeartbeatAt: sql.NullTime{}, |
| 4436 | LastError: pqtype.NullRawMessage{}, |
| 4437 | }) |
| 4438 | if updateErr != nil { |
| 4439 | p.logger.Error(ctx, "failed to release chat acquired during shutdown", |
| 4440 | slog.F("chat_id", chat.ID), slog.Error(updateErr)) |
| 4441 | } |
| 4442 | } |
| 4443 | releaseCancel() |
| 4444 | return |
| 4445 | } |
| 4446 | |
| 4447 | p.inflightMu.Lock() |
| 4448 | for _, chat := range chats { |
| 4449 | p.inflight.Add(1) |
| 4450 | go func() { |
| 4451 | defer p.inflight.Done() |
| 4452 | p.processChat(ctx, chat) |
no test coverage detected