(ticket *flight.Ticket)
| 408 | } |
| 409 | if e.ownsClient { |
| 410 | return e.client.Close() |
| 411 | } |
| 412 | return nil |
| 413 | } |
| 414 | |
| 415 | // Log implements the server query-log forwarding hook without making query |
| 416 | // completion wait on worker RPC or DuckLake writes. |
| 417 | func (e *FlightExecutor) Log(entry wire.QueryLogEntry) { |
| 418 | if e == nil { |
| 419 | return |
| 420 | } |
| 421 | if e.dead.Load() { |
| 422 | observe.AddQueryLogDroppedEntries("forward_worker_dead", 1) |
| 423 | return |
| 424 | } |
| 425 | |
| 426 | e.queryLogMu.Lock() |
| 427 | if e.queryLogClosed { |
| 428 | e.queryLogMu.Unlock() |
| 429 | observe.AddQueryLogDroppedEntries("forward_closed", 1) |
| 430 | return |
| 431 | } |
| 432 | if e.dead.Load() { |
| 433 | e.queryLogMu.Unlock() |
| 434 | observe.AddQueryLogDroppedEntries("forward_worker_dead", 1) |
| 435 | return |
| 436 | } |
| 437 | if e.client == nil || e.client.Client == nil { |
| 438 | e.queryLogMu.Unlock() |
| 439 | observe.AddQueryLogDroppedEntries("forward_unavailable", 1) |
| 440 | return |
| 441 | } |
| 442 | limiter := e.queryLogLimiter |
| 443 | if limiter == nil { |
| 444 | limiter = NewQueryLogLimiter() |
| 445 | e.queryLogLimiter = limiter |
| 446 | } |
| 447 | if !limiter.tryAcquire() { |
| 448 | e.queryLogMu.Unlock() |
| 449 | observe.AddQueryLogDroppedEntries("forward_in_flight_limit", 1) |
| 450 | return |
| 451 | } |
| 452 | baseCtx := context.Background() |
| 453 | if e.ctx != nil { |
| 454 | baseCtx = e.ctx |
| 455 | } |
| 456 | ctx, cancel := context.WithTimeout(baseCtx, queryLogForwardTimeout) |
| 457 | e.queryLogWG.Add(1) |
| 458 | e.queryLogMu.Unlock() |
| 459 | |
| 460 | go func() { |
| 461 | defer e.queryLogWG.Done() |
| 462 | defer limiter.release() |
| 463 | defer cancel() |
| 464 | _ = e.forwardQueryLogEntry(ctx, entry) |
| 465 | }() |
| 466 | } |
| 467 |
no test coverage detected