RollbackDrain restores the session to the state captured in snap, undoing a previous DrainTxLimitedTxn whose frames could not be transmitted. Any data enqueued in the meantime is preserved (appended after the restored bytes). Calling with a nil snapshot is a no-op.
(snap *DrainSnapshot)
| 416 | // enqueued in the meantime is preserved (appended after the restored bytes). |
| 417 | // Calling with a nil snapshot is a no-op. |
| 418 | func (s *Session) RollbackDrain(snap *DrainSnapshot) { |
| 419 | if snap == nil { |
| 420 | return |
| 421 | } |
| 422 | s.mu.Lock() |
| 423 | // Merge: snapshot bytes (drained but unsent) first, then any new bytes |
| 424 | // queued during the in-flight window. |
| 425 | if len(snap.txBuf) > 0 { |
| 426 | if len(s.txBuf) == 0 { |
| 427 | s.txBuf = snap.txBuf |
| 428 | } else { |
| 429 | merged := make([]byte, 0, len(snap.txBuf)+len(s.txBuf)) |
| 430 | merged = append(merged, snap.txBuf...) |
| 431 | merged = append(merged, s.txBuf...) |
| 432 | s.txBuf = merged |
| 433 | } |
| 434 | } |
| 435 | if snap.synNeeded { |
| 436 | s.synNeeded = true |
| 437 | } |
| 438 | // txSeq must reset so retransmitted frames carry the same seq numbers the |
| 439 | // server would have seen on the first (lost) attempt. |
| 440 | s.txSeq = snap.txSeq |
| 441 | if !snap.finSent { |
| 442 | s.finSent = false |
| 443 | s.finSentAt = time.Time{} |
| 444 | } |
| 445 | if !snap.firstQueuedAt.IsZero() { |
| 446 | if s.firstQueuedAt.IsZero() || snap.firstQueuedAt.Before(s.firstQueuedAt) { |
| 447 | s.firstQueuedAt = snap.firstQueuedAt |
| 448 | } |
| 449 | } |
| 450 | cb := s.OnTx |
| 451 | s.txCond.Broadcast() |
| 452 | s.mu.Unlock() |
| 453 | if cb != nil { |
| 454 | cb() |
| 455 | } |
| 456 | } |
| 457 | |
| 458 | // ProcessRx enqueues f to the per-session rxLoop goroutine. The fast path is |
| 459 | // non-blocking. If rxInbox is saturated (slow SOCKS consumer or large burst), |