| 225 | } |
| 226 | |
| 227 | func (s *session) handleAck(body []byte) { |
| 228 | var payload struct { |
| 229 | Seq uint64 `json:"seq"` |
| 230 | } |
| 231 | if len(body) == 0 { |
| 232 | return |
| 233 | } |
| 234 | if err := json.Unmarshal(body, &payload); err != nil { |
| 235 | s.cfg.Logger.Warn("nodeagent: bad ack body", "err", err) |
| 236 | return |
| 237 | } |
| 238 | s.mu.Lock() |
| 239 | ch, ok := s.acks[payload.Seq] |
| 240 | s.mu.Unlock() |
| 241 | if !ok { |
| 242 | return |
| 243 | } |
| 244 | // 用一次性关闭通知所有 WaitAck（理论只一个）。 |
| 245 | select { |
| 246 | case <-ch: |
| 247 | // already closed |
| 248 | default: |
| 249 | close(ch) |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | // shutdown 取消所有 inflight，并等待 dispatch goroutine 退出。 |
| 254 | func (s *session) shutdown() { |