| 442 | // together. |
| 443 | logStuckQuery(snap.session, |
| 444 | withQueryIDAttr([]any{ |
| 445 | "session", snap.key, "rows_processed", pr.rows, "total_rows", pr.total, |
| 446 | "stalled_checks", stallCheckThreshold, |
| 447 | }, snap.queryID)...) |
| 448 | } |
| 449 | } |
| 450 | } |
| 451 | sessionProgress[snap.key] = sessionProgressInfo{Pct: pr.pct, Rows: pr.rows, Total: pr.total, Stalled: stalled} |
| 452 | |
| 453 | case <-time.After(queryProgressTimeout): |
| 454 | // CGO call blocked — DuckDB is busy with I/O (e.g., httpfs |
| 455 | // download). Report as untrackable and don't advance stall |
| 456 | // counters so a legitimately slow operation doesn't get |
| 457 | // flagged as stuck. |
| 458 | sessionProgress[snap.key] = sessionProgressInfo{Pct: -1} |
| 459 | } |
| 460 | } |
| 461 | |
| 462 | // An invalidated instance reports healthy=false as well as the specific |
| 463 | // flag: the flag is what the CP acts and alerts on, while healthy=false |
| 464 | // makes the generic reuse gate (validateReservedWorkerHealth) reject the |
| 465 | // worker even on a CP that predates the flag. |
| 466 | resp, _ := json.Marshal(map[string]interface{}{ |
| 467 | "healthy": !instanceInvalidated, |
| 468 | "draining": h.pool.IsDraining(), |
| 469 | "sessions": h.pool.ActiveSessions(), |
| 470 | "active_queries": h.pool.ActiveDrainWork(), |
| 471 | "uptime_ns": time.Since(h.pool.startTime).Nanoseconds(), |
| 472 | "session_progress": sessionProgress, |