session.go raw
1 package tunnel
2
3 import (
4 "errors"
5 "fmt"
6 "sort"
7 "sync"
8 "sync/atomic"
9 "time"
10
11 "git.smesh.lol/nostrino/pkg/crypto/secp256k1"
12 "git.smesh.lol/nostrino/pkg/nip44"
13 "git.smesh.lol/nostrino/pkg/nostr"
14 "git.smesh.lol/nostrino/pkg/relay"
15 )
16
17 type outFrame struct {
18 data []byte
19 sent time.Time
20 tries int
21 }
22
23 type peerClosedError struct{ reason string }
24
25 func (e peerClosedError) Error() string {
26 if e.reason == "" {
27 return "tunnel: closed by peer"
28 }
29 return "tunnel: closed by peer: " + e.reason
30 }
31
32 type peerResetError struct{ reason string }
33
34 func (e peerResetError) Error() string {
35 if e.reason == "" {
36 return "tunnel: reset by peer"
37 }
38 return "tunnel: reset by peer: " + e.reason
39 }
40
41 type session struct {
42 side side
43 relayMu sync.Mutex
44 relays []*relay.Client
45 key [32]byte
46 localPub [32]byte
47 peerPub [32]byte
48 conv [32]byte
49 id string
50 chain string
51
52 window int
53 heartbeat time.Duration
54 idle time.Duration
55 logf func(string, ...any)
56
57 // wireMu serialises event construction and relay writes so that one
58 // session's frames do not interleave mid-frame at the transport.
59 wireMu sync.Mutex
60
61 // stateMu guards the sequence state and the send window.
62 stateMu sync.Mutex
63 sendSeq uint32
64 unacked map[uint32]*outFrame
65 inflight int
66 peerAck uint32
67 sendCredit int
68 ackCh chan struct{}
69
70 // receive state
71 recvNext uint32
72 recvBuf map[uint32][]byte
73 gapAt time.Time
74
75 mtuLen int64 // negotiated plaintext frame size, atomic
76
77 inbox chan *envelope
78 writeCh chan []byte
79
80 // held is the number of received frames the application has not consumed yet: the
81 // frames parked in recvBuf waiting for a gap to fill, plus the frames in readQueue. It
82 // is an atomic rather than a count under readMu because credit() must see both, and a
83 // second lock taken while readMu is held could invert with pushRead.
84 held int32
85
86 // read queue
87 readMu sync.Mutex
88 readQueue [][]byte
89 readBytes int
90 readWake chan struct{}
91 ackWake chan struct{}
92
93 // deadlines
94 dlMu sync.Mutex
95 readDeadline time.Time
96 writeDeadline time.Time
97 readDLWake chan struct{}
98 writeDLWake chan struct{}
99
100 lastRecv int64 // unix nanos, atomic
101 lastSend int64 // unix nanos, atomic
102
103 done chan struct{}
104 closeOnce sync.Once
105 errMu sync.Mutex
106 err error
107
108 established chan struct{}
109 estOnce sync.Once
110
111 subs []*relay.Subscription
112 onClose func()
113 }
114
115 func newSession(sd side, key [32]byte, localPub, peerPub [32]byte, id string, cfg DialConfig) (*session, error) {
116 if !secp256k1.ValidatePubKey(peerPub) {
117 return nil, errors.New("tunnel: peer key is not on the curve")
118 }
119 conv, err := nip44.ConversationKey(key, peerPub)
120 if err != nil {
121 return nil, fmt.Errorf("tunnel: conversation key: %w", err)
122 }
123 mtu := cfg.MTU
124 if mtu == 0 {
125 mtu = defaultMTU
126 }
127 window := cfg.Window
128 if window == 0 {
129 window = defaultWindow
130 }
131 heartbeat := cfg.Heartbeat
132 if heartbeat == 0 {
133 heartbeat = defaultHeartbeat
134 }
135 idle := cfg.IdleTimeout
136 if idle == 0 {
137 idle = defaultIdle
138 }
139 logf := cfg.Logf
140 if logf == nil {
141 logf = func(string, ...any) {}
142 }
143 s := &session{
144 side: sd,
145 key: key,
146 localPub: localPub,
147 peerPub: peerPub,
148 conv: conv,
149 id: id,
150 chain: cfg.Chain,
151 window: window,
152 heartbeat: heartbeat,
153 idle: idle,
154 logf: logf,
155 unacked: make(map[uint32]*outFrame),
156 sendCredit: window,
157 recvNext: 1,
158 recvBuf: make(map[uint32][]byte),
159 ackCh: make(chan struct{}, 1),
160 inbox: make(chan *envelope, inboxFrames),
161 writeCh: make(chan []byte, sendQueueFrames),
162 readWake: make(chan struct{}, 1),
163 ackWake: make(chan struct{}, 1),
164 readDLWake: make(chan struct{}),
165 writeDLWake: make(chan struct{}),
166 done: make(chan struct{}),
167 established: make(chan struct{}),
168 }
169 atomic.StoreInt64(&s.mtuLen, int64(mtu))
170 atomic.StoreInt64(&s.lastRecv, time.Now().UnixNano())
171 return s, nil
172 }
173
174 func (s *session) mtu() int { return int(atomic.LoadInt64(&s.mtuLen)) }
175
176 func (s *session) start() {
177 go s.sendLoop()
178 go s.recvLoop()
179 go s.ackLoop()
180 if s.side == sideClient {
181 for _, sub := range s.subs {
182 sub := sub
183 go s.clientReadLoop(sub)
184 }
185 }
186 }
187
188 // setRelays installs the relays this session publishes on. The client owns all
189 // of them; the gateway shares them with its listener.
190 func (s *session) setRelays(relays []*relay.Client, subs []*relay.Subscription) {
191 s.relayMu.Lock()
192 s.relays = append([]*relay.Client(nil), relays...)
193 s.subs = append([]*relay.Subscription(nil), subs...)
194 s.relayMu.Unlock()
195 }
196
197 func (s *session) addRelay(rc *relay.Client) bool {
198 s.relayMu.Lock()
199 defer s.relayMu.Unlock()
200 for _, existing := range s.relays {
201 if existing == rc {
202 return false
203 }
204 }
205 s.relays = append(s.relays, rc)
206 return true
207 }
208
209 func (s *session) removeRelay(rc *relay.Client) {
210 s.relayMu.Lock()
211 defer s.relayMu.Unlock()
212 for i, existing := range s.relays {
213 if existing == rc {
214 s.relays = append(s.relays[:i], s.relays[i+1:]...)
215 return
216 }
217 }
218 }
219
220 func (s *session) relayList() []*relay.Client {
221 s.relayMu.Lock()
222 defer s.relayMu.Unlock()
223 return append([]*relay.Client(nil), s.relays...)
224 }
225
226 func (s *session) localKind() int {
227 if s.side == sideGateway {
228 return KindGateway
229 }
230 return KindClient
231 }
232
233 func (s *session) buildEvent(env *envelope) (*nostr.Event, error) {
234 raw, err := env.marshal()
235 if err != nil {
236 return nil, err
237 }
238 content, err := nip44.Encrypt(s.conv, raw)
239 if err != nil {
240 return nil, fmt.Errorf("tunnel: encrypt: %w", err)
241 }
242 return &nostr.Event{
243 Kind: s.localKind(),
244 CreatedAt: time.Now().Unix(),
245 Tags: nostr.Tags{
246 {"p", nostr.Hex(s.peerPub[:])},
247 {"s", s.id},
248 {"enc", "nip44-v2"},
249 {"v", "1"},
250 },
251 Content: content,
252 }, nil
253 }
254
255 func (s *session) sendEnvelope(env *envelope) error {
256 ev, err := s.buildEvent(env)
257 if err != nil {
258 return err
259 }
260 s.wireMu.Lock()
261 defer s.wireMu.Unlock()
262 relays := s.relayList()
263 if len(relays) == 0 {
264 return errors.New("tunnel: no relay")
265 }
266 sent := 0
267 var lastErr error
268 for _, rc := range relays {
269 if err := rc.Send(ev); err != nil {
270 lastErr = err
271 s.removeRelay(rc)
272 s.logf("tunnel: relay send failed: %v", err)
273 continue
274 }
275 sent++
276 }
277 if sent == 0 {
278 return fmt.Errorf("tunnel: relay send: %w", lastErr)
279 }
280 atomic.StoreInt64(&s.lastSend, time.Now().UnixNano())
281 return nil
282 }
283
284 func (s *session) sendHelloAck() {
285 env := &envelope{
286 V: protocolVersion, S: s.id, T: frameHelloAck,
287 MTU: s.mtu(), Win: s.window, PV: protocolVersion, Chain: s.chain,
288 }
289 if err := s.sendEnvelope(env); err != nil {
290 s.closeWithErr(err)
291 }
292 }
293
294 func (s *session) handleHelloAck(env *envelope) error {
295 if env.PV != protocolVersion {
296 return fmt.Errorf("tunnel: gateway protocol version %d", env.PV)
297 }
298 if env.Chain != "" && s.chain != "" && env.Chain != s.chain {
299 return fmt.Errorf("tunnel: gateway chain %q, want %q", env.Chain, s.chain)
300 }
301 if env.MTU >= minMTU && env.MTU < s.mtu() {
302 atomic.StoreInt64(&s.mtuLen, int64(env.MTU))
303 }
304 if env.Win >= minWindow && env.Win < s.window {
305 s.window = env.Win
306 }
307 s.stateMu.Lock()
308 if s.sendCredit > s.window {
309 s.sendCredit = s.window
310 }
311 s.stateMu.Unlock()
312 if env.Chain != "" {
313 s.chain = env.Chain
314 }
315 s.estOnce.Do(func() { close(s.established) })
316 return nil
317 }
318
319 func (s *session) clientReadLoop(sub *relay.Subscription) {
320 for {
321 select {
322 case <-s.done:
323 return
324 case <-sub.Done():
325 // One relay may fail while another still carries the stream. Only
326 // close when the session has no subscriptions left.
327 s.relayMu.Lock()
328 remaining := 0
329 for _, other := range s.subs {
330 if other != sub {
331 select {
332 case <-other.Done():
333 default:
334 remaining++
335 }
336 }
337 }
338 s.relayMu.Unlock()
339 if remaining == 0 {
340 s.closeWithErr(errors.New("tunnel: all relay subscriptions ended"))
341 }
342 return
343 case ev := <-sub.Events:
344 if ev == nil || ev.Kind != KindGateway || ev.PubKey != s.peerPub {
345 continue
346 }
347 raw, err := nip44.Decrypt(s.conv, ev.Content)
348 if err != nil {
349 continue
350 }
351 env, err := parseEnvelope(raw)
352 if err != nil || env.S != s.id {
353 continue
354 }
355 s.enqueue(env)
356 }
357 }
358 }
359
360 func (s *session) enqueue(env *envelope) {
361 select {
362 case s.inbox <- env:
363 default:
364 // The receiver is behind. Dropping a data frame is safe: the gap
365 // triggers a NACK and a retransmission. Dropping a control frame is
366 // recovered by the heartbeat.
367 }
368 }
369
370 func (s *session) recvLoop() {
371 for {
372 select {
373 case <-s.done:
374 return
375 case env := <-s.inbox:
376 atomic.StoreInt64(&s.lastRecv, time.Now().UnixNano())
377 var err error
378 switch env.T {
379 case frameData:
380 err = s.handleData(env)
381 case frameAck:
382 s.handleAck(env)
383 case frameNack:
384 err = s.handleNack(env.Nack)
385 case framePing:
386 err = s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: framePong, Nonce: env.Nonce})
387 case framePong:
388 case frameHelloAck:
389 if s.side == sideClient {
390 err = s.handleHelloAck(env)
391 }
392 case frameHello:
393 // Duplicate HELLO; the gateway answered the original.
394 case frameClose:
395 s.closeWithErr(peerClosedError{env.Reason})
396 return
397 case frameReset:
398 s.resetWithErr(peerResetError{env.Reason})
399 return
400 default:
401 err = fmt.Errorf("tunnel: unknown frame type %q", env.T)
402 }
403 if err != nil {
404 s.closeWithErr(err)
405 return
406 }
407 }
408 }
409 }
410
411 func (s *session) handleData(env *envelope) error {
412 if env.Seq == 0 {
413 return errors.New("tunnel: data frame with sequence zero")
414 }
415 var deliver [][]byte
416 var ack uint32
417 var sendAck bool
418 s.stateMu.Lock()
419 switch {
420 case env.Seq == s.recvNext:
421 deliver = append(deliver, env.Data)
422 // This frame arrived in order and was never buffered, so it is newly held.
423 atomic.AddInt32(&s.held, 1)
424 s.recvNext++
425 for {
426 d, ok := s.recvBuf[s.recvNext]
427 if !ok {
428 break
429 }
430 // Drained frames were counted when they were buffered.
431 delete(s.recvBuf, s.recvNext)
432 deliver = append(deliver, d)
433 s.recvNext++
434 }
435 s.gapAt = time.Time{}
436 ack = s.recvNext - 1
437 sendAck = true
438 case seqBefore(env.Seq, s.recvNext):
439 ack = s.recvNext - 1
440 sendAck = true
441 default:
442 // The defensive bound: with correct credit a sender cannot have more than
443 // window frames unacknowledged, so no frame can be a whole window ahead of
444 // what this side expects. This used to allow window*2, which papered over a
445 // credit bug and then failed at 2*window+1.
446 if env.Seq-s.recvNext >= uint32(s.window) {
447 ahead := env.Seq - s.recvNext
448 s.stateMu.Unlock()
449 return fmt.Errorf(
450 "%w: frame %d is %d ahead of %d, the window is %d",
451 ErrFrameAhead, env.Seq, ahead, s.recvNext, s.window,
452 )
453 }
454 if _, exists := s.recvBuf[env.Seq]; !exists {
455 s.recvBuf[env.Seq] = env.Data
456 atomic.AddInt32(&s.held, 1)
457 }
458 now := time.Now()
459 nack := false
460 if s.gapAt.IsZero() {
461 s.gapAt = now
462 } else if now.Sub(s.gapAt) >= nackDelay {
463 s.gapAt = now
464 nack = true
465 }
466 missing := missingSeqs(s.recvNext, env.Seq, 32)
467 s.stateMu.Unlock()
468 if nack {
469 return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameNack, Nack: missing})
470 }
471 return nil
472 }
473 s.stateMu.Unlock()
474
475 for _, d := range deliver {
476 s.pushRead(d)
477 }
478 if sendAck {
479 return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameAck, Ack: ack, Win: int(s.credit())})
480 }
481 return nil
482 }
483
484 // pushRead queues delivered bytes for Read.
485 //
486 // The frame is counted as held by handleData when it arrives — once, whether it arrived in
487 // order or was parked in recvBuf and drained after a gap filled. read takes the count back
488 // down when the application consumes it. The count is per frame, not per byte, because credit
489 // is a frame budget.
490 func (s *session) pushRead(data []byte) {
491 s.readMu.Lock()
492 s.readQueue = append(s.readQueue, data)
493 s.readBytes += len(data)
494 s.readMu.Unlock()
495 select {
496 case s.readWake <- struct{}{}:
497 default:
498 }
499 }
500
501 // credit is the number of frames the application has room for. It is carried
502 // in every acknowledgement; the sender may not send new data without credit.
503 // A frozen worker therefore stops the bridge's send side and the node's TCP
504 // window takes over, rather than the browser queuing filters without bound.
505 //
506 // It counts every frame the application has not consumed, not just the ones
507 // already in readQueue: a frame held in recvBuf waiting for a gap has arrived
508 // and occupies application room just the same, and a receiver that forgot it
509 // would advertise room for frames it is already holding.
510 func (s *session) credit() uint32 {
511 free := s.window - int(atomic.LoadInt32(&s.held))
512 if free < 0 {
513 free = 0
514 }
515 if free > s.window {
516 free = s.window
517 }
518 return uint32(free)
519 }
520
521 func (s *session) ackLoop() {
522 for {
523 select {
524 case <-s.done:
525 return
526 case <-s.ackWake:
527 s.stateMu.Lock()
528 ack := s.recvNext - 1
529 s.stateMu.Unlock()
530 if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameAck, Ack: ack, Win: int(s.credit())}); err != nil {
531 s.closeWithErr(err)
532 return
533 }
534 }
535 }
536 }
537
538 func (s *session) handleAck(env *envelope) {
539 ack := env.Ack
540 if ack == 0 {
541 return
542 }
543 s.stateMu.Lock()
544 if seqAfter(ack, s.sendSeq) {
545 s.stateMu.Unlock()
546 s.closeWithErr(fmt.Errorf("tunnel: peer acknowledged unsent frame %d", ack))
547 return
548 }
549 newest := seqAfter(ack, s.peerAck)
550 if newest {
551 s.peerAck = ack
552 for seq := range s.unacked {
553 if !seqAfter(seq, ack) {
554 delete(s.unacked, seq)
555 s.inflight--
556 }
557 }
558 if s.inflight < 0 {
559 s.inflight = 0
560 }
561 }
562 // The receiver's free-frame count is authoritative. A stale acknowledgement
563 // must not take credit away, so only the newest cumulative ack applies it.
564 if newest || ack == s.peerAck {
565 // The peer's Win is the room its application has, but this sender's
566 // unacknowledged frames are still in the pipe and still occupy that room. Credit
567 // is therefore the smaller of what the peer advertises and what this side may
568 // still have outstanding; taking Win alone let a sender push a second window
569 // while the first was unanswered.
570 free := s.window - s.inflight
571 if free < 0 {
572 free = 0
573 }
574 wnd := int(env.Win)
575 if wnd > free {
576 wnd = free
577 }
578 if wnd < 0 {
579 wnd = 0
580 }
581 s.sendCredit = wnd
582 }
583 s.stateMu.Unlock()
584 select {
585 case s.ackCh <- struct{}{}:
586 default:
587 }
588 }
589
590 func (s *session) handleNack(seqs []uint32) error {
591 type resend struct {
592 seq uint32
593 data []byte
594 }
595 var out []resend
596 s.stateMu.Lock()
597 for _, seq := range seqs {
598 if f := s.unacked[seq]; f != nil {
599 f.tries++
600 f.sent = time.Now()
601 out = append(out, resend{seq: seq, data: f.data})
602 }
603 }
604 s.stateMu.Unlock()
605 for _, r := range out {
606 ack := s.currentAck()
607 if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: r.seq, Ack: ack, Win: int(s.credit()), Data: r.data}); err != nil {
608 return err
609 }
610 }
611 return nil
612 }
613
614 func (s *session) currentAck() uint32 {
615 s.stateMu.Lock()
616 defer s.stateMu.Unlock()
617 if s.recvNext == 0 {
618 return 0
619 }
620 return s.recvNext - 1
621 }
622
623 func (s *session) sendLoop() {
624 ticker := time.NewTicker(250 * time.Millisecond)
625 defer ticker.Stop()
626 for {
627 select {
628 case <-s.done:
629 return
630 case chunk := <-s.writeCh:
631 if err := s.sendData(chunk); err != nil {
632 s.closeWithErr(err)
633 return
634 }
635 case <-s.ackCh:
636 case now := <-ticker.C:
637 if err := s.tick(now); err != nil {
638 s.closeWithErr(err)
639 return
640 }
641 }
642 }
643 }
644
645 func (s *session) sendData(chunk []byte) error {
646 if err := s.acquireCredit(); err != nil {
647 return err
648 }
649 s.stateMu.Lock()
650 s.sendSeq++
651 seq := s.sendSeq
652 s.unacked[seq] = &outFrame{data: chunk, sent: time.Now(), tries: 1}
653 s.inflight++
654 ack := uint32(0)
655 if s.recvNext > 0 {
656 ack = s.recvNext - 1
657 }
658 s.stateMu.Unlock()
659 if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: seq, Ack: ack, Win: int(s.credit()), Data: chunk}); err != nil {
660 s.stateMu.Lock()
661 delete(s.unacked, seq)
662 s.inflight--
663 s.sendCredit++
664 s.stateMu.Unlock()
665 return err
666 }
667 return nil
668 }
669
670 // acquireCredit takes one frame of receive credit, waiting for the peer to
671 // acknowledge and advertise room. A sender never has more than the negotiated
672 // window of unacknowledged data in the relay.
673 func (s *session) acquireCredit() error {
674 for {
675 s.stateMu.Lock()
676 if s.sendCredit > 0 {
677 s.sendCredit--
678 s.stateMu.Unlock()
679 return nil
680 }
681 s.stateMu.Unlock()
682 select {
683 case <-s.ackCh:
684 case <-s.done:
685 return s.Err()
686 }
687 }
688 }
689
690 func (s *session) tick(now time.Time) error {
691 // Idle check: a stream that has heard nothing for the idle bound is dead
692 // even if the local write side is still buffered.
693 lastRecv := time.Unix(0, atomic.LoadInt64(&s.lastRecv))
694 if s.idle > 0 && now.Sub(lastRecv) > s.idle {
695 return fmt.Errorf("tunnel: idle for %s: %w", now.Sub(lastRecv).Round(time.Second), ErrTimeout)
696 }
697 // Reap frames the peer has not acknowledged.
698 var resend []struct {
699 seq uint32
700 data []byte
701 }
702 dead := false
703 s.stateMu.Lock()
704 for seq, f := range s.unacked {
705 if now.Sub(f.sent) < retransmitAfter {
706 continue
707 }
708 if f.tries >= maxFrameRetries {
709 dead = true
710 break
711 }
712 f.tries++
713 f.sent = now
714 resend = append(resend, struct {
715 seq uint32
716 data []byte
717 }{seq, f.data})
718 }
719 s.stateMu.Unlock()
720 if dead {
721 return errors.New("tunnel: frame retry limit reached")
722 }
723 sort.Slice(resend, func(i, j int) bool { return seqBefore(resend[i].seq, resend[j].seq) })
724 for _, r := range resend {
725 if err := s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frameData, Seq: r.seq, Ack: s.currentAck(), Win: int(s.credit()), Data: r.data}); err != nil {
726 return err
727 }
728 }
729 lastSend := time.Unix(0, atomic.LoadInt64(&s.lastSend))
730 if s.heartbeat > 0 && now.Sub(lastSend) >= s.heartbeat {
731 return s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: framePing, Nonce: uint64(now.UnixNano())})
732 }
733 return nil
734 }
735
736 func (s *session) write(p []byte) (int, error) {
737 total := 0
738 for total < len(p) {
739 n := len(p) - total
740 if mtu := s.mtu(); n > mtu {
741 n = mtu
742 }
743 chunk := make([]byte, n)
744 copy(chunk, p[total:total+n])
745 if err := s.queue(chunk); err != nil {
746 return total, err
747 }
748 total += n
749 }
750 return total, nil
751 }
752
753 func (s *session) queue(chunk []byte) error {
754 for {
755 select {
756 case s.writeCh <- chunk:
757 return nil
758 case <-s.done:
759 return s.Err()
760 default:
761 }
762 dl := s.getWriteDeadline()
763 var timer *time.Timer
764 var tc <-chan time.Time
765 if !dl.IsZero() {
766 remaining := time.Until(dl)
767 if remaining <= 0 {
768 return ErrTimeout
769 }
770 timer = time.NewTimer(remaining)
771 tc = timer.C
772 }
773 wake := s.getWriteWake()
774 select {
775 case s.writeCh <- chunk:
776 stopTimer(timer)
777 return nil
778 case <-s.done:
779 stopTimer(timer)
780 return s.Err()
781 case <-wake:
782 stopTimer(timer)
783 continue
784 case <-tc:
785 return ErrTimeout
786 }
787 }
788 }
789
790 func stopTimer(t *time.Timer) {
791 if t != nil {
792 t.Stop()
793 }
794 }
795
796 func (s *session) read(p []byte) (int, error) {
797 if len(p) == 0 {
798 return 0, nil
799 }
800 for {
801 s.readMu.Lock()
802 if len(s.readQueue) > 0 {
803 chunk := s.readQueue[0]
804 n := copy(p, chunk)
805 if n < len(chunk) {
806 s.readQueue[0] = chunk[n:]
807 } else {
808 s.readQueue[0] = nil
809 s.readQueue = s.readQueue[1:]
810 }
811 s.readBytes -= n
812 consumed := n == len(chunk)
813 s.readMu.Unlock()
814 if consumed {
815 // The application has taken this whole frame, so its room comes back.
816 atomic.AddInt32(&s.held, -1)
817 select {
818 case s.ackWake <- struct{}{}:
819 default:
820 }
821 }
822 return n, nil
823 }
824 s.readMu.Unlock()
825
826 dl := s.getReadDeadline()
827 var timer *time.Timer
828 var tc <-chan time.Time
829 if !dl.IsZero() {
830 remaining := time.Until(dl)
831 if remaining <= 0 {
832 return 0, ErrTimeout
833 }
834 timer = time.NewTimer(remaining)
835 tc = timer.C
836 }
837 wake := s.getReadWake()
838 select {
839 case <-s.readWake:
840 stopTimer(timer)
841 case <-s.done:
842 stopTimer(timer)
843 return 0, s.Err()
844 case <-wake:
845 stopTimer(timer)
846 case <-tc:
847 return 0, ErrTimeout
848 }
849 }
850 }
851
852 func (s *session) setReadDeadline(t time.Time) {
853 s.dlMu.Lock()
854 s.readDeadline = t
855 close(s.readDLWake)
856 s.readDLWake = make(chan struct{})
857 s.dlMu.Unlock()
858 }
859
860 func (s *session) setWriteDeadline(t time.Time) {
861 s.dlMu.Lock()
862 s.writeDeadline = t
863 close(s.writeDLWake)
864 s.writeDLWake = make(chan struct{})
865 s.dlMu.Unlock()
866 }
867
868 func (s *session) getReadDeadline() time.Time {
869 s.dlMu.Lock()
870 defer s.dlMu.Unlock()
871 return s.readDeadline
872 }
873
874 func (s *session) getWriteDeadline() time.Time {
875 s.dlMu.Lock()
876 defer s.dlMu.Unlock()
877 return s.writeDeadline
878 }
879
880 func (s *session) getReadWake() <-chan struct{} {
881 s.dlMu.Lock()
882 defer s.dlMu.Unlock()
883 return s.readDLWake
884 }
885
886 func (s *session) getWriteWake() <-chan struct{} {
887 s.dlMu.Lock()
888 defer s.dlMu.Unlock()
889 return s.writeDLWake
890 }
891
892 // Err returns why the stream ended, or nil while it is open.
893 func (s *session) Err() error {
894 s.errMu.Lock()
895 defer s.errMu.Unlock()
896 if s.err != nil {
897 return s.err
898 }
899 return ErrClosed
900 }
901
902 func (s *session) closeWithErr(err error) { s.closeWith(err, frameClose) }
903
904 func (s *session) resetWithErr(err error) { s.closeWith(err, frameReset) }
905
906 // closeWith ends the session. A local Close sends CLOSE (orderly) or RESET
907 // (abort); a session already ended by the peer or by a transport error does
908 // not send another frame. RESET is the cancellation primitive: the caller is
909 // abandoning the stream, so no in-flight range matters any more.
910 func (s *session) closeWith(err error, frame string) {
911 s.closeOnce.Do(func() {
912 if err == nil {
913 err = ErrClosed
914 }
915 s.errMu.Lock()
916 s.err = err
917 s.errMu.Unlock()
918
919 var closed peerClosedError
920 var reset peerResetError
921 if !errors.As(err, &closed) && !errors.As(err, &reset) && !errors.Is(err, ErrClosed) {
922 _ = s.sendEnvelope(&envelope{V: protocolVersion, S: s.id, T: frame, Reason: err.Error()})
923 }
924 close(s.done)
925 s.relayMu.Lock()
926 subs := append([]*relay.Subscription(nil), s.subs...)
927 relays := append([]*relay.Client(nil), s.relays...)
928 s.relayMu.Unlock()
929 for _, sub := range subs {
930 sub.Close()
931 }
932 if s.side == sideClient {
933 for _, rc := range relays {
934 rc.Close()
935 }
936 }
937 if s.onClose != nil {
938 s.onClose()
939 }
940 })
941 }
942
943 func seqBefore(a, b uint32) bool { return int32(a-b) < 0 }
944 func seqAfter(a, b uint32) bool { return int32(a-b) > 0 }
945
946 func missingSeqs(from, to uint32, max int) []uint32 {
947 var out []uint32
948 for seq := from; seqBefore(seq, to) && len(out) < max; seq++ {
949 out = append(out, seq)
950 }
951 return out
952 }
953